use ::ahash::{AHashMap, AHasher};
use ::anyhow::{ensure, Context as _, Result};
use ::arc_swap::ArcSwapOption;
use ::async_trait::async_trait;
use ::core::option::Option;
use ::crossbeam::queue::SegQueue;
use ::futures::stream::StreamExt as _;
use ::futures::{future, future::BoxFuture, Future, FutureExt, Stream, TryStreamExt};
use ::fxhash::{FxHashMap, FxHashSet, FxHasher64};
use ::indexmap::IndexMap;
use ::log::{debug, error, info, warn};
use ::once_cell::sync::{Lazy, OnceCell};
use ::senax_common::cache::db_cache::{CacheVal, HashVal};
use ::senax_common::cache::msec::MSec;
use ::senax_common::cache::{calc_mem_size, CycleCounter};
use ::senax_common::ShardId;
use ::senax_common::{err, types::blob::*, types::geo_point::*, types::point::*, SqlColumns};
use ::serde_json::Value;
use ::sqlx::query::Query;
use ::sqlx::{Execute as _, FromRow as _};
use ::std::borrow::Borrow;
use ::std::boxed::Box;
use ::std::collections::{BTreeMap, HashMap};
use ::std::convert::TryInto;
use ::std::fmt::Write;
use ::std::hash::{Hash, Hasher};
use ::std::path::Path;
use ::std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use ::std::sync::Arc;
use ::std::time::{SystemTime, UNIX_EPOCH};
use ::std::vec::Vec;
use ::std::{cmp, fmt};
use ::tokio::sync::{mpsc, Mutex, RwLock, Semaphore};
use ::tokio::time::{sleep, Duration};
use ::zstd::{decode_all, encode_all};
@%- if !config.force_disable_cache %@

use ::db::cache::Cache;
@%- else %@
@% endif %@
use ::db::connection::{DbArguments, DbConn, DbRow, DbType};
use crate::misc::{ToBindableJson as _};
use crate::repositories::CacheOpTr;
use ::db::misc::{BindValue, Count, Exists, Updater, Size, TrashMode, UpdaterForInner as _};
use ::db::models::USE_FAST_CACHE;
use ::db::{accessor::*, CacheMsg, BULK_INSERT_MAX_SIZE, IN_CONDITION_LIMIT};
pub use ::db::models::@{ group_name|snake|ident }@::@{ mod_name|ident }@::*;
pub use ::_base_filter_@{ db|snake }@_@{ unified_filter_group }@::repositories::@{ group_name|snake|ident }@::_base::_@{ mod_name }@::*;
use ::_base_filter_@{ db|snake }@_@{ unified_filter_group }@ as _base_filter_;
use ::_base_filter_@{ db|snake }@_@{ unified_filter_group }@::misc::{FilterTr, OrderTr};
@%- if !config.exclude_from_domain %@
#[allow(unused_imports)]
use ::domain::value_objects;
pub use ::domain::repository::@{ db|snake|ident }@::@{ group_name|snake|ident }@::@{ mod_name|ident }@::{join, Joiner_};
@%- for (name, rel_def) in def.belongs_to_outer_db(Joinable::Filter) %@
use ::db_@{ rel_def.db()|snake }@::models::@{ rel_def.get_group_mod_path() }@ as rel_@{ rel_def.get_group_mod_name() }@;
@%- endfor %@
@%- for (name, rel_def) in def.belongs_to_outer_db(Joinable::Join) %@
@%- if !config.exclude_from_domain %@
use ::domain::repository::@{ rel_def.db()|snake|ident }@::@{ rel_def.get_group_mod_path() }@ as join_@{ rel_def.get_group_mod_name() }@;
@%- else %@
use ::db_@{ rel_def.db()|snake }@::repository::@{ rel_def.get_group_name()|snake|ident }@::_base::_@{ rel_def.get_mod_name()|snake }@ as join_@{ rel_def.get_group_mod_name() }@;
@%- endif %@
use _repo_@{ rel_def.db()|snake }@_@{ rel_def.get_group_name()|snake }@::repositories::@{ rel_def.get_base_group_mod_path() }@ as repo_@{ rel_def.get_group_mod_name() }@;
@%- endfor %@
@%- endif %@
@%- for mod_name in def.relation_mods(Joinable::Filter) %@
use ::db::models::@{ mod_name[0]|ident }@::@{ mod_name[1]|ident }@ as rel_@{ mod_name[0] }@_@{ mod_name[1] }@;
@%- endfor %@
@%- for mod_name in def.relation_mods(Joinable::Join) %@
@%- if unified_group.is_ref(mod_name) %@
use ::_base_repo_@{ db|snake }@_@{ UnifiedGroup::unified_name_from_rel(unified_groups, mod_name) }@::repositories::@{ mod_name[0]|ident }@::_base::_@{ mod_name[1] }@ as repo_@{ mod_name[0] }@_@{ mod_name[1] }@;
use ::_base_repo_@{ db|snake }@_@{ UnifiedGroup::unified_name_from_rel(unified_groups, mod_name) }@::repositories::CacheOpTr as _;
@%- else %@
use crate::repositories::@{ mod_name[0]|ident }@::_base::_@{ mod_name[1] }@ as repo_@{ mod_name[0] }@_@{ mod_name[1] }@;
@%- endif %@
@%- if !config.exclude_from_domain %@
use ::domain::repository::@{ db|snake|ident }@::@{ mod_name[0]|ident }@::@{ mod_name[1]|ident }@ as join_@{ mod_name[0] }@_@{ mod_name[1] }@;
@%- else %@
use crate::repositories::@{ mod_name[0]|ident }@::_base::_@{ mod_name[1] }@ as join_@{ mod_name[0] }@_@{ mod_name[1] }@;
@%- endif %@
@%- endfor %@

@% if !config.force_disable_cache -%@
static ALL_ROWS_CACHE: OnceCell<Vec<ArcSwapOption<Vec<_@{ pascal_name }@Cache>>>> = OnceCell::new();
@% else -%@
static ALL_ROWS_CACHE: OnceCell<Vec<ArcSwapOption<Vec<()>>>> = OnceCell::new();
@% endif -%@
static CACHE_RESET_SYNC: OnceCell<Vec<RwLock<u64>>> = OnceCell::new();
static CACHE_RESET_SYNC_ALL_ROWS: OnceCell<Vec<Mutex<u64>>> = OnceCell::new();
static BULK_FETCH_QUEUE: OnceCell<Vec<SegQueue<InnerPrimary>>> = OnceCell::new();
@%- if def.act_as_job_queue() %@
pub static QUEUE_NOTIFIER: tokio::sync::Notify = tokio::sync::Notify::const_new();
@%- endif %@

type __Updater__ = _@{pascal_name}@Updater;

#[allow(clippy::needless_if)]
pub(crate) async fn init() -> Result<()> {
    if ALL_ROWS_CACHE.get().is_none() {
        ALL_ROWS_CACHE.set(DbConn::shard_num_range().map(|_| ArcSwapOption::const_empty()).collect()).unwrap();
        CACHE_RESET_SYNC.set(DbConn::shard_num_range().map(|_| RwLock::new(0)).collect()).unwrap();
        CACHE_RESET_SYNC_ALL_ROWS.set(DbConn::shard_num_range().map(|_| Mutex::new(0)).collect()).unwrap();
        @%- if def.enable_delayed_save() %@
        DELAYED_SAVE_QUEUE.set(DbConn::shard_num_range().map(|_| SegQueue::new()).collect()).unwrap();
        @%- endif %@
        @%- if def.enable_delayed_update() %@
        DELAYED_UPDATE_QUEUE.set(DbConn::shard_num_range().map(|_| SegQueue::new()).collect()).unwrap();
        @%- endif %@
        @%- if def.enable_delayed_upsert() %@
        DELAYED_UPSERT_QUEUE.set(DbConn::shard_num_range().map(|_| SegQueue::new()).collect()).unwrap();
        @%- endif %@
        BULK_FETCH_QUEUE.set(DbConn::shard_num_range().map(|_| SegQueue::new()).collect()).unwrap();
    }

    if !db::is_test_mode() {
        @%- if def.enable_delayed_insert() %@
        tokio::spawn(async {
            while !db::is_stopped() {
                DelayedActor::handle(DelayedMsg::InsertFromDisk);
                sleep(Duration::from_secs(10)).await;
            }
        });
        @%- endif %@
    }
    Ok(())
}

pub(crate) async fn check(shard_id: ShardId) -> Result<()> {
    use anyhow::Context;
    let mut conn = DbConn::_new(shard_id);
    _repo_::query().limit(0).select(&mut conn).await.with_context(|| format!("Schema mismatch detected in the {TABLE_NAME} model on shard {shard_id}."))?;
    Ok(())
}

pub(crate) async fn init_db(path: &Path) -> Result<()> {
    @%- if def.enable_delayed_insert() %@
    let path = path.join(TABLE_NAME);
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_secs(2)).await;
            let db = sled::open(&path);
            match db {
                Ok(db) => {
                    DELAYED_INSERT_DB.set(db).unwrap();
                    DelayedActor::handle(DelayedMsg::InsertFromMemory);
                    DelayedActor::handle(DelayedMsg::InsertFromDisk);
                    break;
                }
                Err(e) => ::log::error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{}", e),
            }
        }
    });
    @%- endif %@
    Ok(())
}
@%- if !config.force_disable_cache %@

impl CacheOpTr<CacheOp, OpData, Data, CacheWrapper, CacheData, PrimaryHasher> for CacheOp {
    @%- if !def.clear_all_cache_on_update() && !def.act_as_job_queue() %@
    fn apply_to_obj(obj: &Option<Arc<CacheWrapper>>, msgs: &[CacheOp], shard_id: ShardId, time: MSec) -> Option<Arc<CacheWrapper>> {
        let mut obj = obj.as_ref().cloned();
        for msg in msgs {
            match msg {
                CacheOp::None => {},
                CacheOp::Insert { data, .. } => {
                    obj = Some(Arc::new(CacheWrapper::from_data(data.clone(), shard_id, time)));
                }
                CacheOp::BulkInsert { .. } => {},
                CacheOp::Update { update, op, .. } => {
                    if let Some(ref arc_wrapper) = obj {
                        let mut wrapper = arc_wrapper.as_ref().clone();
                        wrapper._inner = Self::update(wrapper._inner, update, op);
                        obj = Some(Arc::new(wrapper));
                    } else {
                        obj = None;
                    }
                }
                CacheOp::UpdateMany { .. } => {},
                CacheOp::BulkUpsert { .. } => {},
                CacheOp::Delete { .. } => {
                    obj = None;
                }
                CacheOp::DeleteMany { .. } => {},
                CacheOp::DeleteAll { .. } => {},
                CacheOp::Cascade { .. } => {},
                CacheOp::Invalidate { .. } => {},
                CacheOp::InvalidateAll => {},
            }
        }
        obj
    }

    fn apply_to_list(list: &[Arc<CacheWrapper>], msgs: &[CacheOp], shard_id: ShardId, time: MSec) -> Vec<Arc<CacheWrapper>> {
        let mut map = list.iter().map(|v| (InnerPrimary::from(&v._inner), Arc::clone(v))).collect::<IndexMap<_, _>>();
        for msg in msgs {
            match msg {
                CacheOp::None => {},
                CacheOp::Insert { data, .. } => {
                    map.insert(InnerPrimary::from(data), Arc::new(CacheWrapper::from_data(data.clone(), shard_id, time)));
                }
                CacheOp::BulkInsert { .. } => {},
                CacheOp::Update { id, update, op, .. } => {
                    if let Some(obj) = map.get(id) {
                        let mut wrapper = obj.as_ref().clone();
                        wrapper._inner = Self::update(wrapper._inner, update, op);
                        map.insert(id.clone(), Arc::new(wrapper));
                    }
                }
                CacheOp::UpdateMany { ids, update, data_list, op, .. } => {
                    let mut data_map = FxHashMap::default();
                    for data in data_list.iter() {
                        data_map.insert(InnerPrimary::from(data), data);
                    }
                    for id in ids {
                        if let Some(obj) = map.get(id) {
                            let mut wrapper = obj.as_ref().clone();
                            wrapper._inner = if let Some(data) = data_map.get(id) { 
                                Self::update(wrapper._inner, data, op)
                            } else {
                                Self::update(wrapper._inner, update, op)
                            };
                            map.insert(id.clone(), Arc::new(wrapper));
                        }
                    }
                }
                CacheOp::BulkUpsert { .. } => {},
                CacheOp::Delete { id, .. } => {
                    map.swap_remove(id);
                }
                CacheOp::DeleteMany { ids, .. } => {
                    for id in ids {
                        map.swap_remove(id);
                    }
                }
                CacheOp::DeleteAll { .. } => {},
                CacheOp::Cascade { .. } => {},
                CacheOp::Invalidate { .. } => {},
                CacheOp::InvalidateAll => {},
            }
        }
        map.into_values().collect()
    }

    #[allow(clippy::let_and_return)]
    async fn update_with_unique_cache(id: &PrimaryHasher, obj: CacheData, update: &Data, op: &OpData, time: MSec) -> CacheData {
@%- for (index_name, index) in def.unique_index() %@
        if @{ index.fields(index_name, def)|fmt_index_col_not_null_or_null("op.{ident} != Op::None", "op.{ident} != Op::None && obj.{ident}.is_some()", " && ") }@ {
            let key = VecColKey(vec![@{- index.fields(index_name, def)|fmt_index_col_not_null_or_null("ColKey_::{ident}(obj.{ident}.clone(){inner_to_raw}.into())", "ColKey_::{ident}(obj.{ident}.as_ref().unwrap().clone(){inner_to_raw}.into())", ", ") }@]);
            Cache::invalidate(&key, id._shard_id()).await;
        }
@%- endfor %@
        let obj = CacheOp::update(obj, update, op);
@%- for (index_name, index) in def.unique_index() %@
        if @{ index.fields(index_name, def)|fmt_index_col_not_null_or_null("op.{ident} != Op::None", "op.{ident} != Op::None && obj.{ident}.is_some()", " && ") }@ {
            let key = VecColKey(vec![@{- index.fields(index_name, def)|fmt_index_col_not_null_or_null("ColKey_::{ident}(obj.{ident}.clone(){inner_to_raw}.into())", "ColKey_::{ident}(obj.{ident}.as_ref().unwrap().clone(){inner_to_raw}.into())", ", ") }@]);
            Cache::invalidate(&key, id._shard_id()).await;
        }
@%- endfor %@
        obj
    }
    @%- endif %@

    #[allow(unused_parens)]
    #[allow(clippy::redundant_clone)]
    fn handle_cache_msg(self, sync_map: Arc<FxHashMap<ShardId, u64>>) -> BoxFuture<'static, ()> {
        async move {
            let time = MSec::now();
            @%- if !config.force_disable_cache %@
            for (shard_id, sync) in sync_map.iter() {
                if *sync == 0 {
                    let shard_id = *shard_id;
                    tokio::spawn(async move {
                        tokio::time::sleep(Duration::from_secs(5)).await;
                        __clear_cache(shard_id, 0, false).await;
                    });
                }
            }
            @%- endif %@
            match self {
                CacheOp::None => {},
                @%- if def.act_as_job_queue() %@
                CacheOp::Queued => {
                    QUEUE_NOTIFIER.notify_waiters();
                },
                @%- endif %@
                @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && !def.act_as_job_queue() %@
                CacheOp::Insert { overwrite, shard_id, data
                    @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join(", _{rel_name}", "") -}@ 
                    @{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join(", _{rel_name}", "") }@ } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    let mut cache = CacheWrapper::from_data(data.clone(), shard_id, time);
                    let id = InnerPrimary::from(&cache._inner);
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        let op = if overwrite {
                            db::models::NotifyOp::upsert
                        } else {
                            db::models::NotifyOp::insert
                        };
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, op, &id).await;
                    }
                    let id = PrimaryHasher(id, shard_id);
    @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
                    if let Some(_{rel_name}) = _{rel_name} {
                        for msg in _{rel_name} {
                            msg.handle_cache_msg(Arc::clone(&sync_map)).await;
                        }
                    }", "")|replace1("_inner") }@
    @{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("
                    if let Some(_{rel_name}) = _{rel_name} {
                        for msg in _{rel_name} {
                            msg.handle_cache_msg(Arc::clone(&sync_map)).await;
                        }
                    }", "")|replace1("_inner") }@
                    if overwrite {
                        let cs = CacheSyncWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            sync,
                        };
                        Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                        Cache::invalidate(&id, shard_id).await;
                    }
                    @%- for (index_name, index) in def.unique_index() %@
                    if @{ index.fields(index_name, def)|fmt_index_col_not_null_or_null("true", " data.{ident}.is_some()", " && ") }@ {
                        let key = VecColKey(vec![@{- index.fields(index_name, def)|fmt_index_col_not_null_or_null("ColKey_::{ident}(data.{ident}.clone(){inner_to_raw}.into())", "ColKey_::{ident}(data.{ident}.unwrap().clone(){inner_to_raw}.into())", ", ") }@]);
                        Cache::invalidate(&key, shard_id).await;
                    }
                    @%- endfor %@
                }
                CacheOp::BulkInsert { replace, overwrite, ignore, shard_id, list } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    let op = if replace || overwrite {
                        db::models::NotifyOp::upsert
                    } else {
                        db::models::NotifyOp::insert
                    };
                    for row in list {
                        let mut cache = CacheWrapper::from_data(row._data.clone(), shard_id, time);
                        let id = InnerPrimary::from(&cache._inner);
                        if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, op, &id).await;
                        }
                        let id = PrimaryHasher(id, shard_id);
                        @%- if def.versioned %@
                        let vw = VersionWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            version: cache._inner.@{ ConfigDef::version()|ident }@,
                        };
                        if let Some(old) = Cache::get_version::<VersionWrapper>(&vw, shard_id).await.filter(|o| o.id == id.0) {
                            if old.version.less_than(vw.version) {
                                Cache::insert_version(&vw, Arc::new(vw.clone())).await;
                            }
                        } else if !row._is_new {
                            Cache::insert_version(&vw, Arc::new(vw.clone())).await;
                        }
                        @%- endif %@
                        let cs = CacheSyncWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            sync,
                        };
                        Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                        Cache::invalidate(&id, shard_id).await;
                        @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
                        if let Some(Some(_{rel_name})) = row.{rel_name} {
                            rel_{class_mod}::CacheOp::BulkInsert {
                                replace,
                                overwrite,
                                ignore, 
                                shard_id,
                                list: vec![*_{rel_name}],
                            }.handle_cache_msg(Arc::clone(&sync_map)).await;
                        }", "")|replace1("_data") }@
                        @{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("
                        if let Some(_{rel_name}) = row.{rel_name} {
                            rel_{class_mod}::CacheOp::BulkInsert {
                                replace,
                                overwrite,
                                ignore, 
                                shard_id,
                                list: _{rel_name},
                            }.handle_cache_msg(Arc::clone(&sync_map)).await;
                        }", "")|replace1("_data") }@
                        @%- for (index_name, index) in def.unique_index() %@
                        if @{ index.fields(index_name, def)|fmt_index_col_not_null_or_null("true", " row._data.{ident}.is_some()", " && ") }@ {
                            let key = VecColKey(vec![@{- index.fields(index_name, def)|fmt_index_col_not_null_or_null("ColKey_::{ident}(row._data.{ident}.clone(){inner_to_raw}.into())", "ColKey_::{ident}(row._data.{ident}.unwrap().clone(){inner_to_raw}.into())", ", ") }@]);
                            Cache::invalidate(&key, shard_id).await;
                        }
                        @%- endfor %@
                    }
                    @%- if def.has_auto_primary() %@
                    if overwrite {
                        __clear_cache(shard_id, sync, false).await;
                        if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::invalidate_all, &serde_json::Value::Null).await;
                        }
                    }
                    @%- endif %@
                }
                CacheOp::Update { id, shard_id, update, op
                    @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join(", _{rel_name}", "") -}@ 
                    @{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join(", _{rel_name}", "") }@ } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::update, &id).await;
                    }
                    if USE_CACHE {
                        let id = PrimaryHasher(id.clone(), shard_id);
                        @%- if def.versioned %@
                        let vw = VersionWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            version: update.@{ version_col }@,
                        };
                        if let Some(old) = Cache::get_version::<VersionWrapper>(&vw, shard_id).await.filter(|o| o.id == id.0) {
                            if old.version.less_than(vw.version) {
                                Cache::insert_version(&vw, Arc::new(vw.clone())).await;
                            } else {
                                return;
                            }
                        } else {
                            Cache::insert_version(&vw, Arc::new(vw.clone())).await;
                        }
                        @%- endif %@
                        if let Some(cache) = Cache::get_from_memory::<CacheWrapper>(&id, shard_id, USE_FAST_CACHE).await.filter(|o| InnerPrimary::from(o) == id.0) {
                            let mut cache = cache.as_ref().clone();
                            @%- if def.versioned %@
                            if op.@{ version_col }@ == Op::Set {
                                if cache._inner.@{ version_col }@.greater_equal(update.@{ version_col }@) {
                                    return;
                                } else if cache._inner.@{ version_col }@.cycle_add(1).less_than(update.@{ version_col }@) {
                                    Cache::invalidate(&id, shard_id).await;
                                    return;
                                }
                            }
                            @%- endif %@
                            cache._inner = CacheOp::update_with_unique_cache(&id, cache._inner, &update, &op, time).await;
                            @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
                            if let Some(_{rel_name}) = _{rel_name} {
                                if let Some(rel) = &mut cache.{rel_name} {
                                    *rel = rel_{class_mod}::CacheOp::apply_to_obj(rel, &_{rel_name}, shard_id, time){soft_delete_filter};
                                }
                                for msg in _{rel_name} {
                                    msg.handle_cache_msg(Arc::clone(&sync_map)).await;
                                }
                            }", "")|replace1("_inner") }@
                            @{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("
                            if let Some(_{rel_name}) = _{rel_name} {
                                if let Some(rel) = &mut cache.{rel_name} {
                                    *rel = rel_{class_mod}::CacheOp::apply_to_list(rel, &_{rel_name}, shard_id, time).into_iter(){soft_delete_filter}.collect();
                                    {cache_list_sort}
                                    {cache_list_limit}
                                }
                                for msg in _{rel_name} {
                                    msg.handle_cache_msg(Arc::clone(&sync_map)).await;
                                }
                            }", "")|replace1("_inner") }@
                            Cache::insert_long(&id, Arc::new(cache), USE_FAST_CACHE).await;
                        } else {
                            let cs = CacheSyncWrapper {
                                id: id.0.clone(),
                                shard_id,
                                time,
                                sync,
                            };
                            Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                            Cache::invalidate(&id, shard_id).await;
                        }
                    }
                }
                CacheOp::UpdateMany { ids, shard_id, update, data_list, op } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        for id in &ids {
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::update, id).await;
                        }
                    }
                    if USE_CACHE {
                        let mut data_map = FxHashMap::default();
                        for data in data_list.into_iter() {
                            data_map.insert(PrimaryHasher(InnerPrimary::from(&data), shard_id), data);
                        }
                        for id in &ids {
                            let id = PrimaryHasher(id.clone(), shard_id);
                            if let Some(cache) = Cache::get_from_memory::<CacheWrapper>(&id, shard_id, USE_FAST_CACHE).await.filter(|o| InnerPrimary::from(o) == id.0) {
                                let mut cache = cache.as_ref().clone();
                                cache._inner = if let Some(data) = data_map.get(&id) {
                                    CacheOp::update_with_unique_cache(&id, cache._inner, data, &op, time).await
                                } else {
                                    CacheOp::update_with_unique_cache(&id, cache._inner, &update, &op, time).await
                                };
                                Cache::insert_long(&id, Arc::new(cache), USE_FAST_CACHE).await;
                            } else {
                                let cs = CacheSyncWrapper {
                                    id: id.0.clone(),
                                    shard_id,
                                    time,
                                    sync,
                                };
                                Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                                Cache::invalidate(&id, shard_id).await;
                            }
                        }
                    }
                }
                CacheOp::BulkUpsert { shard_id, data_list, update, op } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        for data in &data_list {
                            let id = InnerPrimary::from(data);
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::upsert, &id).await;
                        }
                    }
                    if USE_CACHE {
                        for data in &data_list {
                            let id = PrimaryHasher(InnerPrimary::from(data), shard_id);
                            if let Some(cache) = Cache::get_from_memory::<CacheWrapper>(&id, shard_id, USE_FAST_CACHE).await.filter(|o| InnerPrimary::from(o) == id.0) {
                                let mut cache = cache.as_ref().clone();
                                cache._inner = CacheOp::update_with_unique_cache(&id, cache._inner, &update, &op, time).await;
                                Cache::insert_long(&id, Arc::new(cache), USE_FAST_CACHE).await;
                            } else {
                                let cs = CacheSyncWrapper {
                                    id: id.0.clone(),
                                    shard_id,
                                    time,
                                    sync,
                                };
                                Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                                Cache::invalidate(&id, shard_id).await;
                            }
                        }
                    }
                }
                CacheOp::Delete { id, shard_id } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::delete, &id).await;
                    }
                    if USE_CACHE {
                        let id = PrimaryHasher(id.clone(), shard_id);
                        @%- if def.versioned %@
                        let vw = VersionWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            version: 0,
                        };
                        Cache::invalidate_version(&id, shard_id).await;
                        @%- endif %@
                        let cs = CacheSyncWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            sync,
                        };
                        Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                        Cache::invalidate(&id, shard_id).await;
                    }
                }
                CacheOp::DeleteMany { ids, shard_id } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        for id in &ids {
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::delete, id).await;
                        }
                    }
                    if USE_CACHE {
                        for id in &ids {
                            let id = PrimaryHasher(id.clone(), shard_id);
                            @%- if def.versioned %@
                            let vw = VersionWrapper {
                                id: id.0.clone(),
                                shard_id,
                                time,
                                version: 0,
                            };
                            Cache::invalidate_version(&id, shard_id).await;
                            @%- endif %@
                            let cs = CacheSyncWrapper {
                                id: id.0.clone(),
                                shard_id,
                                time,
                                sync,
                            };
                            Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                            Cache::invalidate(&id, shard_id).await;
                        }
                    }
                }
                CacheOp::DeleteAll { shard_id } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    __clear_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::delete_all, &serde_json::Value::Null).await;
                    }
                }
                CacheOp::Cascade { ids, shard_id } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        for id in &ids {
                            DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::delete, id).await;
                        }
                    }
                    if USE_CACHE {
                        for id in &ids {
                            let id = PrimaryHasher(id.clone(), shard_id);
                            @%- if def.versioned %@
                            let vw = VersionWrapper {
                                id: id.0.clone(),
                                shard_id,
                                time,
                                version: 0,
                            };
                            Cache::invalidate_version(&id, shard_id).await;
                            @%- endif %@
                            let cs = CacheSyncWrapper {
                                id: id.0.clone(),
                                shard_id,
                                time,
                                sync,
                            };
                            Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                            Cache::invalidate(&id, shard_id).await;
                        }
                    }
                }
                CacheOp::Invalidate { id, shard_id  } => {
                    let sync = *sync_map.get(&shard_id).unwrap();
                    clear_all_rows_cache(shard_id, sync, false).await;
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::invalidate, &id).await;
                    }
                    if USE_CACHE {
                        let id = PrimaryHasher(id.clone(), shard_id);
                        @%- if def.versioned %@
                        let vw = VersionWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            version: 0,
                        };
                        Cache::invalidate_version(&id, shard_id).await;
                        @%- endif %@
                        let cs = CacheSyncWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time,
                            sync,
                        };
                        Cache::insert_version(&cs, Arc::new(cs.clone())).await;
                        Cache::invalidate(&id, shard_id).await;
                    }
                }
                @%- endif %@
                CacheOp::InvalidateAll => {
                    for (shard_id, sync) in sync_map.iter() {
                        __clear_cache(*shard_id, *sync, false).await;
                    }
                    if ENABLE_UPDATE_NOTICE && DbConn::_has_update_notice() {
                        DbConn::_push_update_notice(db::models::TableName::@{ table_name|ident }@, db::models::NotifyOp::invalidate_all, &serde_json::Value::Null).await;
                    }
                }
            }
        }.boxed()
    }
}
@%- endif %@
@%- if !config.force_disable_cache %@

async fn clear_all_rows_cache(shard_id: ShardId, sync: u64, clear_test: bool) {
    @{- def.auto_inc_or_seq()|fmt_join("
    if clear_test {
        if let Some(ids) = GENERATED_IDS.get() { ids.write().unwrap().clear() }
    }", "") }@
    if ENABLE_ALL_ROWS_CACHE {
        if let Some(list) = CACHE_RESET_SYNC_ALL_ROWS.get() {
            if clear_test {
                for shard in list {
                    let mut _sync = shard.lock().await;
                    *_sync = 0;
                }
            }
            let mut _sync = list[shard_id as usize].lock().await;
            if sync == 0 || sync > *_sync {
                *_sync = sync;
                let _ = ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].swap(None);
            }
        }
    }
}
@%- endif %@
pub(crate) async fn __clear_cache(_shard_id: ShardId, _sync: u64, _clear_test: bool) {
    @%- if !config.force_disable_cache %@
    clear_all_rows_cache(_shard_id, _sync, _clear_test).await;
    if USE_CACHE {
        if let Some(s) = CACHE_RESET_SYNC.get() {
            let mut sync = s[_shard_id as usize].write().await;
            *sync = _sync;
            Cache::invalidate_all_of::<CacheWrapper>();
            Cache::invalidate_all_of::<PrimaryWrapper>();
            Cache::invalidate_all_of_version::<VersionWrapper>();
            Cache::invalidate_all_of_version::<CacheSyncWrapper>();
        }
    }
    @%- endif %@
}
@% if def.enable_delayed_insert() %@
static DELAYED_INSERT_QUEUE: Lazy<SegQueue<ForInsert>> = Lazy::new(SegQueue::new);
static DELAYED_INSERT_DB: OnceCell<sled::Db> = OnceCell::new();
static DELAYED_INSERT_WAITING: AtomicBool = AtomicBool::new(false);
@%- endif %@
static DELAYED_DB_NO: Lazy<AtomicU64> = Lazy::new(|| {
    let now = SystemTime::now();
    let time = now.duration_since(UNIX_EPOCH).unwrap();
    AtomicU64::new(time.as_secs() << 20)
});
@%- if def.enable_delayed_save() %@
static DELAYED_SAVE_QUEUE: OnceCell<Vec<SegQueue<__Updater__>>> = OnceCell::new();
static DELAYED_SAVE_WAITING: AtomicBool = AtomicBool::new(false);
static DELAYED_SAVE_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(1));
@%- endif %@
@%- if def.enable_delayed_update() %@
static DELAYED_UPDATE_QUEUE: OnceCell<Vec<SegQueue<__Updater__>>> = OnceCell::new();
static DELAYED_UPDATE_WAITING: AtomicBool = AtomicBool::new(false);
static DELAYED_UPDATE_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(1));
@%- endif %@
@%- if def.enable_delayed_upsert() %@
static DELAYED_UPSERT_QUEUE: OnceCell<Vec<SegQueue<__Updater__>>> = OnceCell::new();
static DELAYED_UPSERT_WAITING: AtomicBool = AtomicBool::new(false);
static DELAYED_UPSERT_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(1));
@%- endif %@

struct DelayedActor;
@%- if def.enable_delayed_insert() %@

struct DelayedInsertBuf(Vec<ForInsert>);
impl Drop for DelayedInsertBuf {
    fn drop(&mut self) {
        if !self.0.is_empty() {
            if let Err(err) = push_delayed_db(&self.0) {
                error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "push_delayed_db:{}", err);
            }
        }
    }
}
@%- endif %@

enum DelayedMsg {
    @%- if def.enable_delayed_insert() %@
    InsertFromMemory,
    InsertFromDisk,
    @%- endif %@
    @%- if def.enable_delayed_save() %@
    Save,
    @%- endif %@
    @%- if def.enable_delayed_update() %@
    Update,
    @%- endif %@
    @%- if def.enable_delayed_upsert() %@
    Upsert,
    @%- endif %@
}
impl DelayedActor {
    fn handle(msg: DelayedMsg) {
        match msg {
            @%- if def.enable_delayed_insert() %@
            DelayedMsg::InsertFromMemory => {
                if DELAYED_INSERT_WAITING.load(Ordering::SeqCst) {
                    return;
                }
                tokio::spawn(
                    async move {
                        DELAYED_INSERT_WAITING.store(true, Ordering::SeqCst);
                        let _guard = db::get_shutdown_guard();
                        sleep(Duration::from_millis(100)).await;
                        DELAYED_INSERT_WAITING.store(false, Ordering::SeqCst);
                        handle_delayed_msg_insert_from_memory().await;
                    }
                );
            }
            DelayedMsg::InsertFromDisk => {
                tokio::spawn(
                    async move {
                        let _guard = db::get_shutdown_guard();
                        let mut handles = Vec::new();
                        for shard_id in DbConn::shard_num_range() {
                            handles.push(handle_delayed_msg_insert_from_disk(shard_id));
                        }
                        future::join_all(handles).await.iter().for_each(|r| {
                            if let Err(err) = r {
                                error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED ERROR:{}", err);
                            }
                        });
                    }
                );
            }
            @%- endif %@
            @%- if def.enable_delayed_save() %@
            DelayedMsg::Save => {
                if DELAYED_SAVE_WAITING.load(Ordering::SeqCst) {
                    return;
                }
                tokio::spawn(
                    async move {
                        DELAYED_SAVE_WAITING.store(true, Ordering::SeqCst);
                        let _guard = db::get_shutdown_guard();
                        let _semaphore = DELAYED_SAVE_SEMAPHORE.acquire().await;
                        DELAYED_SAVE_WAITING.store(false, Ordering::SeqCst);
                        handle_delayed_msg_save().await;
                    }
                );
            }
            @%- endif %@
            @%- if def.enable_delayed_update() %@
            DelayedMsg::Update => {
                if DELAYED_UPDATE_WAITING.load(Ordering::SeqCst) {
                    return;
                }
                tokio::spawn(
                    async move {
                        DELAYED_UPDATE_WAITING.store(true, Ordering::SeqCst);
                        let _guard = db::get_shutdown_guard();
                        let _semaphore = DELAYED_UPDATE_SEMAPHORE.acquire().await;
                        DELAYED_UPDATE_WAITING.store(false, Ordering::SeqCst);
                        handle_delayed_msg_update().await;
                    }
                );
            }
            @%- endif %@
            @%- if def.enable_delayed_upsert() %@
            DelayedMsg::Upsert => {
                if DELAYED_UPSERT_WAITING.load(Ordering::SeqCst) {
                    return;
                }
                tokio::spawn(
                    async move {
                        DELAYED_UPSERT_WAITING.store(true, Ordering::SeqCst);
                        let _guard = db::get_shutdown_guard();
                        let _semaphore = DELAYED_UPSERT_SEMAPHORE.acquire().await;
                        DELAYED_UPSERT_WAITING.store(false, Ordering::SeqCst);
                        handle_delayed_msg_upsert().await;
                    }
                );
            }
            @%- endif %@
        }
    }
}
@%- if def.enable_delayed_save() %@

async fn handle_delayed_msg_save() {
    let mut handles = Vec::new();
    for shard_id in DbConn::shard_num_range() {
        handles.push(_handle_delayed_msg_save(shard_id));
    }
    future::join_all(handles).await;
}
@%- endif %@
@%- if def.enable_delayed_update() %@

async fn handle_delayed_msg_update() {
    let mut handles = Vec::new();
    for shard_id in DbConn::shard_num_range() {
        handles.push(_handle_delayed_msg_update(shard_id));
    }
    future::join_all(handles).await;
}
@%- endif %@
@%- if def.enable_delayed_upsert() %@

async fn handle_delayed_msg_upsert() {
    let mut handles = Vec::new();
    for shard_id in DbConn::shard_num_range() {
        handles.push(_handle_delayed_msg_upsert(shard_id));
    }
    future::join_all(handles).await;
}
@%- endif %@
@%- if def.enable_delayed_insert() %@

async fn handle_delayed_msg_insert_from_memory() {
    let mut vec = Vec::with_capacity(DELAYED_INSERT_QUEUE.len() + 10);
    while let Some(x) = DELAYED_INSERT_QUEUE.pop() {
        vec.push(x);
    }
    if vec.is_empty() {
        return;
    }
    let mut conn_list = Vec::new();
    for shard_id in DbConn::shard_num_range() {
        let mut conn = DbConn::_new(shard_id);
        match conn.begin_immediately().await {
            Ok(_) => {
                conn_list.push(conn);
            }
            Err(err) => {
                error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED ERROR:{}", err);
            }
        }
    }
    if conn_list.is_empty() {
        if let Err(err) = push_delayed_db(&vec) {
            error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "push_delayed_db:{}", err);
        }
        return;
    }
    let len = vec.len() / conn_list.len() + 1;
    let mut handles = Vec::new();
    for conn in conn_list {
        if !vec.is_empty() {
            let vec2 = vec.split_off(vec.len().saturating_sub(len));
            handles.push(_handle_delayed_msg_insert_from_memory(conn, vec2));
        }
    }
    future::join_all(handles).await;
}

async fn _handle_delayed_msg_insert_from_memory(mut conn: DbConn, vec: Vec<ForInsert>) {
    let mut buf = DelayedInsertBuf(vec);
    let result = __bulk_insert(&mut conn, &buf.0, true, false, false).await;
    if let Err(err) = result {
        if let Some(err) = err.downcast_ref::<sqlx::Error>() {
            match err {
                sqlx::Error::Io(..) => {
                    // retry all
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED ERROR:{}", err);
                    drop(buf);
                    return;
                }
                sqlx::Error::WorkerCrashed => {
                    // retry all
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED ERROR:{}", err);
                    drop(buf);
                    return;
                }
                _ => {
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED FAILED:{}", err);
                }
            }
        } else {
            error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED FAILED:{}", err);
        }
    }
    let result = conn.commit().await;
    if db::connection::_is_retryable_error(result, TABLE_NAME) {
        drop(buf);
        return;
    }
    buf.0.clear();
}

async fn handle_delayed_msg_insert_from_disk(shard_id: ShardId) -> Result<()> {
    if db::is_stopped() {
        return Ok(());
    }
    let db = match DELAYED_INSERT_DB.get() {
        Some(db) => db,
        None => {
            return Ok(());
        }
    };
    if db.is_empty() {
        return Ok(());
    }
    let mut conn = DbConn::_new(shard_id);
    if let Err(err) = conn.begin_immediately().await {
        error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "INSERT DELAYED ERROR:{}", err);
        return Ok(());
    }
    let mut vec = Vec::new();
    let mut total_size = 0;
    let max_size = *BULK_INSERT_MAX_SIZE.get().unwrap() / 10 * 9;
    while let Ok(x) = db.pop_min() {
        if let Some(x) = x {
            let mut bytes = bytes::Bytes::from(decode_all::<&[u8]>(x.1.borrow())?);
            let list: Vec<ForInsert> = senax_encoder::unpack(&mut bytes)?;
            for data in list {
                total_size += data._data._size();
                vec.push(data);
            }
            if total_size >= max_size {
                break;
            }
        } else {
            break;
        }
    }
    if vec.is_empty() {
        return Ok(());
    }
    _handle_delayed_msg_insert_from_memory(conn, vec).await;
    if db.is_empty() {
        info!("Insert delayed successfully recovered.");
    } else if !db::is_stopped() {
        DelayedActor::handle(DelayedMsg::InsertFromDisk);
    }
    tokio::spawn(async move {
        let _guard = db::get_shutdown_guard();
        let _ = db.flush_async().await;
    });
    Ok(())
}

fn push_delayed_db(list: &Vec<ForInsert>) -> Result<()> {
    if let Some(db) = DELAYED_INSERT_DB.get() {
        let no = DELAYED_DB_NO.fetch_add(1, Ordering::SeqCst);
        let bytes = senax_encoder::pack(&list)?;
        let mut buf = encode_all(bytes.as_ref(), 3)?;
        db.insert(no.to_be_bytes(), buf)?;
        tokio::spawn(async move {
            let _guard = db::get_shutdown_guard();
            let _ = db.flush_async().await;
        });
    } else {
        for data in list {
            DELAYED_INSERT_QUEUE.push(data.clone());
        }
    }
    Ok(())
}
@%- endif %@
@%- if def.enable_delayed_save() %@

async fn _handle_delayed_msg_save(shard_id: ShardId) {
    let mut map: BTreeMap<InnerPrimary, IndexMap<OpData, __Updater__>> = BTreeMap::new();
    while let Some(x) = DELAYED_SAVE_QUEUE.get().unwrap()[shard_id as usize].pop() {
        let inner_map = map.entry(InnerPrimary::from(&x)).or_default();
        if let Some(old) = inner_map.get_mut(&x._op) {
            aggregate_update(&x, old);
        } else {
            inner_map.insert(x._op.clone(), x);
        }
    }
    if map.is_empty() {
        return;
    }
    let mut vec: Vec<IndexMap<OpData, __Updater__>> = map.into_values().collect();
    let chunk_num = cmp::max(10, vec.len() * 2 / (DbConn::max_connections_for_write() as usize) + 1);
    let mut join_set = tokio::task::JoinSet::new();
    while !vec.is_empty() {
        let mut buf = vec.split_off(vec.len().saturating_sub(chunk_num));
        join_set.spawn(async move {
            loop {
                let mut conn = DbConn::_new(shard_id);
                if let Err(err) = conn.begin_immediately().await {
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "SAVE DELAYED ERROR:{}", err);
                    sleep(Duration::from_secs(10)).await;
                    continue;
                }
                for inner_map in buf.iter() {
                    for (_op, updater) in inner_map.iter() {
                        let result = __save(&mut conn, updater.clone(), 0).await;
                        if db::connection::_is_retryable_error(result, TABLE_NAME) {
                            sleep(Duration::from_secs(1)).await;
                            continue;
                        }
                    }
                }
                let result = conn.commit().await;
                if db::connection::_is_retryable_error(result, TABLE_NAME) {
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                break;
            }
        });
    }
    while join_set.join_next().await.is_some() {};
}
@%- endif %@
@%- if def.has_delayed_update() %@

fn aggregate_update(x: &_@{ pascal_name }@Updater, old: &mut _@{ pascal_name }@Updater) {
    @{- def.non_primaries_except_read_only()|fmt_join("
    Accessor{accessor_with_sep_type}::_set(x._op.{ident}, &mut old._update.{ident}, &x._update.{ident});", "") }@
}
@%- endif %@
@%- if def.enable_delayed_update() %@

async fn _handle_delayed_msg_update(shard_id: ShardId) {
    @{- def.soft_delete_tpl2("","
    let deleted_at = Some(SystemTime::now().into());","","
    let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});")}@
    let mut map: BTreeMap<InnerPrimary, IndexMap<OpData, __Updater__>> = BTreeMap::new();
    while let Some(mut x) = DELAYED_UPDATE_QUEUE.get().unwrap()[shard_id as usize].pop() {
        @{- def.soft_delete_tpl2("","
        if x.will_be_deleted() {
            x.mut_deleted_at().set(deleted_at);
        }","
        if x.will_be_deleted() {
            x.mut_deleted().set(true);
        }","
        if x.will_be_deleted() {
            x.mut_deleted().set(deleted);
        }")}@
        let inner_map = map.entry(InnerPrimary::from(&x)).or_default();
        if let Some(old) = inner_map.get_mut(&x._op) {
            aggregate_update(&x, old);
        } else {
            inner_map.insert(x._op.clone(), x);
        }
    }
    if map.is_empty() {
        return;
    }
    let mut vec: Vec<(OpData, Data, Vec<InnerPrimary>)> = Vec::new();
    for m in map.into_values() {
        'next: for up in m.into_values() {
            let id: InnerPrimary = (&up).into();
            for row in vec.iter_mut() {
                if row.0 == up._op && row.1 == up._update {
                    row.2.push(id);
                    continue 'next;
                }
            }
            vec.push((up._op, up._update, vec![id]));
        }
    }
    let mut join_set = tokio::task::JoinSet::new();
    for (op, update, list) in vec {
        join_set.spawn(async move {
            loop {
                let mut conn = DbConn::_new(shard_id);
                if let Err(err) = conn.begin_immediately().await {
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "UPDATE DELAYED ERROR:{}", err);
                    sleep(Duration::from_secs(10)).await;
                    continue;
                }
                let mut updater = __Updater__ {
                    _data: Data::default(),
                    _update: update.clone(),
                    _filter_flag: Default::default(),
                    _is_new: false,
                    _do_delete: false,
                    _upsert: false,
                    _is_loaded: true,
                    _op: op.clone(),
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
                };
                @%- if def.updated_at_conf().is_some() %@
                if updater._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
                    updater.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
                }
                @%- endif %@
                let result = __update_many(&mut conn, list.clone(), updater).await;
                if db::connection::_is_retryable_error(result, TABLE_NAME) {
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                let result = conn.commit().await;
                if db::connection::_is_retryable_error(result, TABLE_NAME) {
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                break;
            }
        });
    }
    while join_set.join_next().await.is_some() {};
}
@%- endif %@
@%- if def.enable_delayed_upsert() %@

async fn _handle_delayed_msg_upsert(shard_id: ShardId) {
    @{- def.soft_delete_tpl2("","
    let deleted_at = Some(SystemTime::now().into());","","
    let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});")}@
    let mut map: BTreeMap<InnerPrimary, IndexMap<OpData, __Updater__>> = BTreeMap::new();
    while let Some(mut x) = DELAYED_UPSERT_QUEUE.get().unwrap()[shard_id as usize].pop() {
        @{- def.soft_delete_tpl2("","
        if x.will_be_deleted() {
            x.mut_deleted_at().set(deleted_at);
        }","
        if x.will_be_deleted() {
            x.mut_deleted().set(true);
        }","
        if x.will_be_deleted() {
            x.mut_deleted().set(deleted);
        }")}@
        let inner_map = map.entry(InnerPrimary::from(&x)).or_default();
        if let Some(old) = inner_map.get_mut(&x._op) {
            aggregate_update(&x, old);
        } else {
            inner_map.insert(x._op.clone(), x);
        }
    }
    if map.is_empty() {
        return;
    }
    let mut vec: Vec<(OpData, Data, Vec<Data>)> = Vec::new();
    for m in map.into_values() {
        'next: for up in m.into_values() {
            for row in vec.iter_mut() {
                if row.0 == up._op && row.1 == up._update {
                    row.2.push(up._data);
                    continue 'next;
                }
            }
            vec.push((up._op, up._update, vec![up._data]));
        }
    }
    let mut join_set = tokio::task::JoinSet::new();
    for (op, update, list) in vec {
        join_set.spawn(async move {
            loop {
                let mut conn = DbConn::_new(shard_id);
                if let Err(err) = conn.begin_immediately().await {
                    error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", table = TABLE_NAME; "UPSERT DELAYED ERROR:{}", err);
                    sleep(Duration::from_secs(10)).await;
                    continue;
                }
                let mut updater = __Updater__ {
                    _data: list[0].clone(),
                    _update: update.clone(),
                    _filter_flag: Default::default(),
                    _is_new: false,
                    _do_delete: false,
                    _upsert: false,
                    _is_loaded: true,
                    _op: op.clone(),
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
                    {rel_name}: None,", "") }@
                };
                @%- if def.updated_at_conf().is_some() %@
                if updater._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
                    updater.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
                }
                @%- endif %@
                let result = __bulk_upsert(&mut conn, &list, &updater).await;
                if db::connection::_is_retryable_error(result, TABLE_NAME) {
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                let result = conn.commit().await;
                if db::connection::_is_retryable_error(result, TABLE_NAME) {
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                break;
            }
        });
    }
    while join_set.join_next().await.is_some() {};
}
@%- endif %@

#[async_trait]
pub trait _@{ pascal_name }@Joiner {
    async fn join(&mut self, conn: &mut DbConn, joiner: Option<Box<Joiner_>>) -> Result<()> {
        if let Some(joiner) = joiner {
            @{- def.relations(Joinable::Join)|fmt_rel_join("
            if joiner.{rel_name}.is_some() {
                self.join_{raw_rel_name}(conn, joiner.{rel_name}).await?;
            }", "") }@
            @{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
            if joiner.{rel_name}.is_some() {
                self.join_{raw_rel_name}(conn, joiner.{rel_name}).await?;
            }", "") }@
        }
        Ok(())
    }
@{- def.relations_one_and_belonging(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()>;", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()>;", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()>;", "") }@
}

#[async_trait]
impl _@{ pascal_name }@Joiner for _@{ pascal_name }@ {
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(conn, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(&mut conn._{db_snake}_db, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select(conn).await?.pop().map(Box::new));
        Ok(())
    }", "") }@
@{- def.relations_many_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let mut l = repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select(conn).await?;
        {list_sort}
        self.{rel_name} = Some(l);
        Ok(())
    }", "") }@
@{- def.relations_many_with_limit(Joinable::Join)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let order = vec![{order}];
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).order_by(order){limit}.select(conn).await?);
        Ok(())
    }", "") }@
}
@%- if !def.disable_update() %@

#[async_trait]
impl _@{ pascal_name }@Joiner for _@{ pascal_name }@Updater {
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(conn, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(&mut conn._{db_snake}_db, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select_for_update(conn).await?);
        Ok(())
    }", "") }@
@{- def.relations_many_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let mut l = repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select_for_update(conn).await?;
        {list_sort_for_update}
        self.{rel_name} = Some(l);
        Ok(())
    }", "") }@
@{- def.relations_many_with_limit(Joinable::Join)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let order = vec![{order}];
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).order_by(order){limit}.select_for_update(conn).await?);
        Ok(())
    }", "") }@
}
@%- endif %@
@%- if !config.force_disable_cache %@
@%- if def.use_cache() %@
trait CacheWrapperTr {
    fn _has_join(&self, joiner: &Option<Box<Joiner_>>) -> bool;
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()>;", "") }@
@{- def.relations_many_cache_without_limit(Joinable::Join)|fmt_rel_join("
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()>;", "") }@
@{- def.relations_many_cache_with_limit(Joinable::Join)|fmt_rel_join("
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()>;", "") }@
}

impl CacheWrapperTr for CacheWrapper {
    fn _has_join(&self, joiner: &Option<Box<Joiner_>>) -> bool {
        if let Some(joiner) = joiner {
@{- def.relations_in_cache(Joinable::Join)|fmt_rel_join("\n            if self.{rel_name}.is_none() && joiner.{rel_name}.is_some() { return false; }", "") }@
        }
        true
    }
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).__select_for_cache(conn).await?.into_iter().map(|v| v._wrapper).next());
        Ok(())
    }", "") }@
@{- def.relations_many_cache_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let mut l: Vec<_> = repo_{class_mod}::_repo_::query().filter(filter).__select_for_cache(conn).await?.into_iter().map(|v| v._wrapper).collect();
        {list_sort}
        self.{rel_name} = Some(l);
        Ok(())
    }", "") }@
@{- def.relations_many_cache_with_limit(Joinable::Join)|fmt_rel_join("
    async fn fetch_{raw_rel_name}(&mut self, conn: &mut DbConn) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let order = vec![{order}];
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).order_by(order){limit}.__select_for_cache(conn).await?.into_iter().map(|v| v._wrapper).collect());
        Ok(())
    }", "") }@
}

#[async_trait]
impl _@{ pascal_name }@Joiner for _@{ pascal_name }@Cache {
@{- def.relations_belonging_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional_from_cache{with_trashed}(conn, id, joiner).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_uncached(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(conn, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if let Some(id) = RelPk{rel_name_pascal}::primary(self) {
            let obj = repo_{class_mod}::_repo_::find_optional{with_trashed}(&mut conn._{db_snake}_db, id, joiner, None, None).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        } else {
            self.{rel_name} = Some(None);
        }
        Ok(())
    }", "") }@
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if !matches!(joiner.as_ref().map(|v| v.has_some()), Some(true)) {
            if self.{rel_name}.is_none() && let Some(v) = self._wrapper.{rel_name}.as_ref() {
                self.{rel_name} = Some(v.clone().map(|v| Box::new((v, Default::default()).into())));
            }
            return Ok(());
        }
        if let Some(obj) = self._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\") {
            let id: rel_{class_mod}::InnerPrimary = obj.into();
            let obj = repo_{class_mod}::_repo_::find_optional_from_cache(conn, &id, joiner).await?;
            if let Some(obj) = obj {
                self.{rel_name} = Some(Some(Box::new(obj)));
            } else {
                self.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_one_uncached(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select(conn).await?.pop().map(Box::new));
        Ok(())
    }", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if !matches!(joiner.as_ref().map(|v| v.has_some()), Some(true)) {
            if self.{rel_name}.is_none() && let Some(v) = self._wrapper.{rel_name}.as_ref() {
                self.{rel_name} = Some(v.iter().map(|v| (v.clone(), Default::default()).into()).collect());
            }
            return Ok(());
        }
        let ids: Vec<rel_{class_mod}::InnerPrimary> = self._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\").iter().map(|v| v.into()).collect();
        if !ids.is_empty() {
            let list = repo_{class_mod}::_repo_::find_many_from_cache(conn, ids.iter(), joiner).await?;
            let mut map: AHashMap<_, _> = list.into_iter().map(|v| (rel_{class_mod}::InnerPrimary::from(&v), v)).collect();
            self.{rel_name} = Some(ids.iter().flat_map(|id| map.remove(id)).collect());
        } else {
            self.{rel_name} = Some(Vec::new());
        }
        Ok(())
    }", "") }@
@{- def.relations_many_uncached_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let mut l = repo_{class_mod}::_repo_::query().filter(filter).join(joiner).select(conn).await?;
        {list_sort}
        self.{rel_name} = Some(l);
        Ok(())
    }", "") }@
@{- def.relations_many_uncached_with_limit(Joinable::Join)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let filter = RelFil{rel_name_pascal}::filter(self){additional_filter};
        let order = vec![{order}];
        self.{rel_name} = Some(repo_{class_mod}::_repo_::query().filter(filter).join(joiner).order_by(order){limit}.select(conn).await?);
        Ok(())
    }", "") }@
}
@%- endif %@
@%- endif %@

#[async_trait]
impl _@{ pascal_name }@Joiner for Vec<_@{ pascal_name }@> {
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(conn, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(&mut conn._{db_snake}_db, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.insert(id, Box::new(row));
                }
            }
        }
        for val in self.iter_mut() {
            val.{rel_name} = Some(map.remove(&(&*val).into()));
        }
        Ok(())
    }", "") }@
@{- def.relations_many_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.entry(id).or_default().push(row);
                }
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_many_with_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let queries: Result<Vec<_>> = self.iter().map(|v| {
            let filter = RelFil{rel_name_pascal}::filter(v){additional_filter};
            anyhow::Ok(repo_{class_mod}::_repo_::query().filter(filter){order_and_limit})
        }).collect();
        let mut list = repo_{class_mod}::UnionBuilder::select_union(queries?, conn, None, None, None).await?;
        repo_{class_mod}::{class}Joiner::join(&mut list, conn, joiner).await?;
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for row in list {
            if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                map.entry(id).or_default().push(row);
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
}
@%- if !def.disable_update() %@

#[async_trait]
impl _@{ pascal_name }@Joiner for Vec<_@{ pascal_name }@Updater> {
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(conn, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(&mut conn._{db_snake}_db, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select_for_update(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._data) {
                    map.entry(id).or_default().push(row);
                }
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort_for_update}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_many_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select_for_update(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._data) {
                    map.entry(id).or_default().push(row);
                }
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort_for_update}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_many_with_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let queries: Result<Vec<_>> = self.iter().map(|v| {
            let filter = RelFil{rel_name_pascal}::filter(v){additional_filter};
            anyhow::Ok(repo_{class_mod}::_repo_::query().filter(filter){order_and_limit})
        }).collect();
        let mut list = repo_{class_mod}::UnionBuilder::select_union_for_update(queries?, conn).await?;
        repo_{class_mod}::{class}Joiner::join(&mut list, conn, joiner).await?;
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for row in list {
            if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._data) {
                map.entry(id).or_default().push(row);
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort_for_update}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
}
@%- endif %@
@%- if !config.force_disable_cache %@
@%- if def.use_cache() %@

trait CacheWrapperVecTr {
    @{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()>;", "") }@
@{- def.relations_many_cache_without_limit(Joinable::Join)|fmt_rel_join("
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()>;", "") }@
@{- def.relations_many_cache_with_limit(Joinable::Join)|fmt_rel_join("
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()>;", "") }@
}

impl CacheWrapperVecTr for CacheWrapper {
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()> {
        if vec.is_empty() { return Ok(()); }
        let mut map = AHashMap::default();
        for chunk in vec.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list: Vec<Arc<_>> = repo_{class_mod}::_repo_::query().filter(filter).__select_for_cache(conn).await?.into_iter().map(|v| v._wrapper).collect();
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.insert(id, row);
                }
            }
        }
        for val in vec.iter_mut() {
            val.{rel_name} = Some(map.remove(&(&*val).into()));
        }
        Ok(())
    }", "") }@
@{- def.relations_many_cache_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()> {
        if vec.is_empty() { return Ok(()); }
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for chunk in vec.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list: Vec<Arc<_>> = repo_{class_mod}::_repo_::query().filter(filter).__select_for_cache(conn).await?.into_iter().map(|v| v._wrapper).collect();
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.entry(id).or_default().push(row);
                }
            }
        }
        for val in vec.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_many_cache_with_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn fetch_{raw_rel_name}_for_vec(vec: &mut [CacheWrapper], conn: &mut DbConn) -> Result<()> {
        if vec.is_empty() { return Ok(()); }
        let queries: Vec<_> = vec.iter().map(|v| {
            let filter = RelFil{rel_name_pascal}::filter(v){additional_filter};
            repo_{class_mod}::_repo_::query().filter(filter){order_and_limit}
        }).collect();
        use rel_{class_mod}::_UnionBuilder;
        let list: Vec<Arc<_>> = queries.__select_union_for_cache(conn).await?.into_iter().map(|v| v._wrapper).collect();
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for row in list {
            if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                map.entry(id).or_default().push(row);
            }
        }
        for val in vec.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
}

#[async_trait]
impl _@{ pascal_name }@Joiner for Vec<_@{ pascal_name }@Cache> {
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if !matches!(joiner.as_ref().map(|v| v.has_some()), Some(true)) {
            for val in self.iter_mut() {
                if val.{rel_name}.is_none() && let Some(v) = val._wrapper.{rel_name}.as_ref() {
                    val.{rel_name} = Some(v.clone().map(|v| Box::new((v, Default::default()).into())));
                }
            }
            return Ok(());
        }
        let ids: Vec<rel_{class_mod}::InnerPrimary> = self.iter().flat_map(|v| v._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\")).map(|v| v.into()).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many_from_cache(conn, ids.iter(), joiner).await?;
        let mut map: AHashMap<_, _> = list.into_iter().map(|v| (rel_{class_mod}::InnerPrimary::from(&v), v)).collect();
        for val in self.iter_mut() {
            if let Some(v) = &val._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\") {
                val.{rel_name} = Some(map.remove(&v.into()).map(Box::new));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_one_uncached(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.insert(id, Box::new(row));
                }
            }
        }
        for val in self.iter_mut() {
            val.{rel_name} = Some(map.remove(&(&*val).into()));
        }
        Ok(())
    }", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if !matches!(joiner.as_ref().map(|v| v.has_some()), Some(true)) {
            for val in self.iter_mut() {
                if val.{rel_name}.is_none() && let Some(v) = val._wrapper.{rel_name}.as_ref() {
                    val.{rel_name} = Some(v.iter().map(|v| (v.clone(), Default::default()).into()).collect());
                }
            }
            return Ok(());
        }
        let ids: Vec<rel_{class_mod}::InnerPrimary> = self.iter().flat_map(|v| v._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\")).map(|v| v.into()).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many_from_cache(conn, ids.iter(), joiner).await?;
        let mut map: AHashMap<_, _> = list.into_iter().map(|v| (rel_{class_mod}::InnerPrimary::from(&v), v)).collect();
        for val in self.iter_mut() {
            val.{rel_name} = Some(val._wrapper.{rel_name}.as_ref().expect(\"{raw_rel_name} is not loaded\").iter().flat_map(|v| map.remove(&v.into())).collect());
        }
        Ok(())
    }", "") }@
@{- def.relations_many_uncached_without_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for chunk in self.chunks(IN_CONDITION_LIMIT) {
            let filter = RelFil{rel_name_pascal}::in_filter(chunk){additional_filter};
            let list = repo_{class_mod}::_repo_::query().filter(filter).join(joiner.clone()).select(conn).await?;
            for row in list {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                    map.entry(id).or_default().push(row);
                }
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_many_uncached_with_limit(Joinable::Join)|fmt_rel_join("
    #[allow(unused_parens)]
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        if self.is_empty() { return Ok(()); }
        let queries: Result<Vec<_>> = self.iter().map(|v| {
            let filter = RelFil{rel_name_pascal}::filter(v){additional_filter};
            anyhow::Ok(repo_{class_mod}::_repo_::query().filter(filter){order_and_limit})
        }).collect();
        let mut list = repo_{class_mod}::UnionBuilder::select_union(queries?, conn, None, None, None).await?;
        repo_{class_mod}::{class}Joiner::join(&mut list, conn, joiner).await?;
        let mut map: AHashMap<_, Vec<_>> = AHashMap::default();
        for row in list {
            if let Some(id) = RelFk{rel_name_pascal}::get_fk(&row._inner) {
                map.entry(id).or_default().push(row);
            }
        }
        for val in self.iter_mut() {
            let mut l = map.remove(&(&*val).into()).unwrap_or_default();
            {list_sort}
            val.{rel_name} = Some(l);
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_cache(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many_from_cache{with_trashed}(conn, ids.iter(), joiner).await?;
        let map = repo_{class_mod}::_repo_::cache_list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_uncached(Joinable::Join, false)|fmt_rel_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(conn, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("
    async fn join_{raw_rel_name}(&mut self, conn: &mut DbConn, joiner: Option<Box<join_{class_mod}::Joiner_>>) -> Result<()> {
        let ids: FxHashSet<_> = self.iter().flat_map(RelPk{rel_name_pascal}::primary).collect();
        if ids.is_empty() { return Ok(()); }
        let list = repo_{class_mod}::_repo_::find_many{with_trashed}(&mut conn._{db_snake}_db, ids.iter(), joiner, None, None).await?;
        let map = repo_{class_mod}::_repo_::list_to_map(list);
        for val in self.iter_mut() {
            if let Some(id) = RelPk{rel_name_pascal}::primary(val) {
                val.{rel_name} = Some(map.get(&id).map(|v| Box::new(v.clone())));
            } else {
                val.{rel_name} = Some(None);
            }
        }
        Ok(())
    }", "") }@
}
@%- endif %@
@%- endif %@

@%- if config.exclude_from_domain %@

#[derive(Debug, Clone, Default)]
pub struct Joiner_ {
@{- def.relations(Joinable::Join)|fmt_rel_join("
    pub {rel_name}: Option<Box<join_{class_mod}::Joiner_>>,", "") }@
    pub _dummy: bool,
}
impl Joiner_ {
    #[allow(clippy::nonminimal_bool)]
    pub fn has_some(&self) -> bool {
        false
        @{- def.relations(Joinable::Join)|fmt_rel_join("
            || self.{rel_name}.is_some()", "") }@
    }
    pub fn merge(lhs: Option<Box<Self>>, rhs: Option<Box<Self>>) -> Option<Box<Self>> {
        if let Some(lhs) = lhs {
            if let Some(rhs) = rhs {
                Some(Box::new(Joiner_{
                    @{- def.relations(Joinable::Join)|fmt_rel_join("
                    {rel_name}: join_{class_mod}::Joiner_::merge(lhs.{rel_name}, rhs.{rel_name}),", "") }@
                    ..Default::default()
                }))
            } else {
                Some(lhs)
            }
        } else {
            rhs
        }
    }
}
@%- let fetch_macro_name = "{}_{}_{}"|format(db|snake, group_name, model_name) %@
@%- let base_path = "$crate::models::{}::{}::_base::_{}"|format(db|snake|ident, group_name|ident, mod_name) %@
@%- let model_path = "$crate::repositories::{}::_base::_{}"|format(group_name|snake|ident, mod_name) -%@
#[macro_export]
macro_rules! _join_@{ fetch_macro_name }@ {
@{- def.relations(Joinable::Join)|fmt_rel_join("
    ({rel_name}) => ($crate::models::{class_mod_path}::_{mod_name}::join!({}));
    ({rel_name}: $p:tt) => ($crate::models::{class_mod_path}::_{mod_name}::join!($p));", "") }@
    () => ();
}
pub use _join_@{ fetch_macro_name }@ as _join;
#[macro_export]
macro_rules! join_@{ fetch_macro_name }@ {
    ({$($i:ident $(: $p:tt)?),*}) => (Some(Box::new(@{ model_path }@::Joiner_ {
        $($i: @{ base_path }@::_join!($i $(: $p)?),)*
        ..Default::default()
    })));
}
pub use join_@{ fetch_macro_name }@ as join;
@%- endif %@

#[derive(Debug, Clone, Default)]
pub struct QueryBuilder {
    filter: Option<Filter_>,
    filter_flag: BTreeMap<&'static str, Filter_>,
    order: Option<Vec<Order_>>,
    raw_order: Option<String>,
    limit: Option<usize>,
    offset: Option<usize>,
    skip_locked: bool,
    trash_mode: TrashMode,
    raw_query: String,
    bind: Vec<BindValue>,
    joiner: Option<Box<Joiner_>>,
}

impl QueryBuilder {
    pub fn filter(mut self, filter: Filter_) -> Self {
        self.filter = Some(filter);
        self
    }
    pub fn with_filter_flag(mut self, filter_flag: BTreeMap<&'static str, Filter_>) -> Self {
        self.filter_flag = filter_flag;
        self
    }
    pub fn order_by(mut self, order: Vec<Order_>) -> Self {
        if order.is_empty() {
            self.order = None;
        } else {
            self.order = Some(order);
        }
        self
    }
    pub fn raw_order_by(mut self, order: &str) -> Self {
        if order.is_empty() {
            self.raw_order = None;
        } else {
            self.raw_order = Some(order.to_string());
        }
        self
    }
    pub fn limit(mut self, limit: usize) -> Self {
        self.limit = Some(limit);
        self
    }
    pub fn offset(mut self, offset: usize) -> Self {
        self.offset = Some(offset);
        self
    }
    pub fn skip_locked(mut self) -> Self {
        self.skip_locked = true;
        self
    }
    @%- if def.is_soft_delete() %@
    pub fn with_trashed(mut self) -> Self {
        self.trash_mode = TrashMode::With;
        self
    }
    pub fn only_trashed(mut self) -> Self {
        self.trash_mode = TrashMode::Only;
        self
    }
    @%- endif %@
    pub fn append_raw_query(mut self, query: &str) -> Self {
        self.raw_query.push_str(query);
        self
    }
    /// bind for raw_query
    pub fn bind<T: Into<BindValue>>(mut self, value: T) -> Self {
        self.bind.push(value.into());
        self
    }
    pub fn when<F>(self, condition: bool, f: F) -> Self
    where
        F: FnOnce(Self) -> Self,
    {
        if condition {
            f(self)
        } else {
            self
        }
    }
    pub fn if_let_some<T, F>(self, value: &Option<T>, f: F) -> Self
    where
        F: FnOnce(Self, &T) -> Self,
    {
        if let Some(v) = value {
            f(self, v)
        } else {
            self
        }
    }
    pub fn join(mut self, joiner: Option<Box<Joiner_>>) -> Self {
        self.joiner = Joiner_::merge(self.joiner, joiner);
        self
    }
    async fn _select<T>(self, conn: &mut DbConn) -> Result<Vec<(T, BTreeMap<&'static str, bool>)>>
    where
        T: for<'r> sqlx::FromRow<'r, DbRow> + SqlColumns + Send + Sync + Unpin,
    {
        let result: sqlx::Result<Vec<_>> = self.__select(T::_sql_cols(@{ is_mysql_str }@), conn).await?.into_iter().map(|(v, f)| T::from_row(&v).map(|v| (v, f))).collect();
        Ok(result?)
    }
    async fn __select(self, sql_cols: &str, conn: &mut DbConn) -> Result<Vec<(DbRow, BTreeMap<&'static str, bool>)>> {
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let filter_flag_names: Vec<_> = self.filter_flag.keys().cloned().collect();
        let sql = self._sql(sql_cols, false, conn.shard_id(), &filter_digest);
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        query = self._bind(query, true);
        let now = std::time::Instant::now();
        let result = crate::misc::fetch!(conn, query, fetch_all);
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        Ok(result
            .into_iter()
            .map(|row| {
                use sqlx::Row;
                let mut flags = BTreeMap::new();
                for name in filter_flag_names.clone() {
                    flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
                }
                (row, flags)
            })
            .collect())
    }

    #[allow(clippy::if_same_then_else)]
    fn _sql(&self, sql_cols: &str, for_update: bool, shard_id: ShardId, filter_digest: &str) -> String {
        let force_indexes = make_force_indexes(filter_digest);
        let sql_cols = write_filter_flag(sql_cols, &self.filter_flag, shard_id);
        let mut sql = format!(
            r#"SELECT {}{} FROM @{ table_name|db_esc }@ as _t1 {} {} {}"#,
            if !force_indexes.is_empty() {
                format!("/*+ INDEX(_t1 {}) */ ", force_indexes.join(","))
            } else {
                String::new()
            },
            sql_cols,
            Filter_::write_where(
                &self.filter,
                self.trash_mode,
                TRASHED_SQL,
                NOT_TRASHED_SQL,
                ONLY_TRASHED_SQL,
                shard_id,
            ),
            &self.raw_query,
            Order_::write_order(&self.order, &self.raw_order),
        );
        if let Some(limit) = self.limit {
            write!(sql, " limit {}", limit).unwrap();
        }
        if let Some(offset) = self.offset {
            write!(sql, " offset {}", offset).unwrap();
        }
        if for_update {
            if self.skip_locked {
                write!(sql, " FOR UPDATE SKIP LOCKED").unwrap();
            } else {
                write!(sql, " FOR UPDATE").unwrap();
            }
        }
        sql
    }

    fn _bind(self, mut query: Query<DbType, DbArguments>, filter_flag: bool) -> Query<DbType, DbArguments> {
        if filter_flag {
            for (_name, filter) in self.filter_flag {
                query = filter.bind_to_query(query);
            }
        }
        if let Some(c) = self.filter {
            debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "filter: {:?}", &c);
            query = c.bind_to_query(query);
        }
        for value in self.bind.into_iter() {
            query = value.bind(query);
        }
        query
    }

    async fn _select_stream(self, sql_cols: &str, conn: &mut DbConn) -> Result<mpsc::Receiver<(DbRow, BTreeMap<&'static str, bool>)>> {
        let ctx_no = conn.ctx_no();
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        let filter_flag_names: Vec<_> = self.filter_flag.keys().cloned().collect();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = ctx_no; "filter digest:{}", filter_digest);
        let sql = self._sql(sql_cols, false, conn.shard_id(), &filter_digest);
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let (tx, rx) = mpsc::channel(1000);
        let mut executor = conn.acquire_reader().await?;
        tokio::spawn(async move {
            let mut query = sqlx::query(&sql);
            debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = ctx_no, sql = &query.sql(); "query");
            query = self._bind(query, true);
            let now = std::time::Instant::now();
            let mut result = query.fetch(executor.as_mut());
            if now.elapsed() > std::time::Duration::from_secs(1) {
                warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = ctx_no; "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
            }
            while let Some(v) = result.try_next().await.unwrap_or_else(|e| {
                warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{}", e);
                None
            }) {
                use sqlx::Row;
                let mut flags = BTreeMap::new();
                for name in filter_flag_names.clone() {
                    flags.insert(name, v.get(name));
                }
                if let Err(e) = tx.send((v, flags)).await {
                    warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{}", e);
                    break;
                }
            }
        });
        Ok(rx)
    }
    @%- if def.use_cache() %@

    #[allow(clippy::if_same_then_else)]
    async fn _select_from_cache(mut self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Cache>> {
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let force_indexes = make_force_indexes(&filter_digest);
        let filter_flag_names: Vec<_> = self.filter_flag.keys().cloned().collect();
        let sql_cols = write_filter_flag(r#"@{ def.primaries()|fmt_join("{col_query}", ", ") }@"#, &self.filter_flag, conn.shard_id());
        let mut sql = format!(
            r#"SELECT {}{sql_cols} FROM @{ table_name|db_esc }@ as _t1 {} {} {}"#,
            if !force_indexes.is_empty() {
                format!("/*+ INDEX(_t1 {}) */ ", force_indexes.join(","))
            } else {
                String::new()
            },
            Filter_::write_where(&self.filter, self.trash_mode, TRASHED_SQL, NOT_TRASHED_SQL, ONLY_TRASHED_SQL, conn.shard_id()),
            &self.raw_query,
            Order_::write_order(&self.order, &self.raw_order),
        );
        if let Some(limit) = self.limit {
            write!(sql, " limit {}", limit)?;
        }
        if let Some(offset) = self.offset {
            write!(sql, " offset {}", offset)?;
        }
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        let joiner = self.joiner.take();
        query = self._bind(query, true);
        let now = std::time::Instant::now();
        let result = crate::misc::fetch!(conn, query, fetch_all);
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        let result: sqlx::Result<Vec<_>> = result
            .iter()
            .map(|row| {
                use sqlx::Row;
                let mut flags: BTreeMap<&str, bool> = BTreeMap::new();
                for name in filter_flag_names.clone() {
                    flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
                }
                InnerPrimary::from_row(row).map(|v| (v, flags))
            })
            .collect();
        let result = result?;
        let ids: Vec<(@{ def.primaries()|fmt_join_with_paren("{outer_owned}", ", ") }@, _)> = result.iter().map(|(id, flags)| (id.into(), flags)).collect();
        conn.release_cache_tx();
        let list = _repo_::find_many_from_cache(conn, ids.iter().map(|v| v.0.clone()), joiner).await?;
        let mut map = _repo_::cache_list_to_map(list);
        let list: Vec<_@{ pascal_name }@Cache> = result
            .into_iter()
            .flat_map(|(ref id, flags)| map.remove(&id.into()).map(|mut v| {v._filter_flag = flags; v}))
            .collect();
        Ok(list)
    }
    @%- endif %@
    @%- if !def.disable_update() %@

    pub async fn select_for_update(mut self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Updater>> {
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let filter_flag_names: Vec<_> = self.filter_flag.keys().cloned().collect();
        let sql = self._sql(Data::_sql_cols(@{ is_mysql_str }@), !conn.wo_tx(), conn.shard_id(), &filter_digest);
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        let joiner = self.joiner.take();
        query = self._bind(query, true);
        let now = std::time::Instant::now();
        let result = if conn.wo_tx() {
            query.fetch_all(conn.acquire_writer().await?.as_mut()).await?
        } else {
            query.fetch_all(conn.get_tx().await?.as_mut()).await?
        };
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        let result: sqlx::Result<Vec<_>> = result.into_iter().map(|row| {
            use sqlx::Row;
            let mut flags = BTreeMap::new();
            for name in filter_flag_names.clone() {
                flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
            }            
            Data::from_row(&row).map(|v| (v, flags))}).collect();
        let mut list: Vec<__Updater__> = result?
            .into_iter()
            .map(__Updater__::from)
            .collect();
        list.join(conn, joiner).await?;
        Ok(list)
    }
    @%- endif %@

    pub async fn select(mut self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@>> {
        let joiner = self.joiner.take();
        let result: Vec<(Data, _)> = self._select(conn).await?;
        #[allow(unused_mut)]
        let mut list: Vec<_@{ pascal_name }@> = result.into_iter().map(_@{ pascal_name }@::from).collect();
        list.join(conn, joiner).await?;
        Ok(list)
    }

    pub async fn select_one(mut self, conn: &mut DbConn) -> Result<Option<_@{ pascal_name }@>> {
        self.limit = Some(1);
        let mut list = Self::select(self, conn).await?;
        Ok(list.pop())
    }

    pub async fn select_for<T>(self, conn: &mut DbConn) -> Result<Vec<T>>
    where
        T: for<'r> sqlx::FromRow<'r, DbRow> + SqlColumns + Send + Sync + Unpin,
    {
        self._select(conn).await.map(|v| v.into_iter().map(|v| v.0).collect())
    }

    pub async fn select_one_for<T>(mut self, conn: &mut DbConn) -> Result<Option<T>>
    where
        T: for<'r> sqlx::FromRow<'r, DbRow> + SqlColumns + Send + Sync + Unpin,
    {
        self.limit = Some(1);
        let mut list = Self::select_for(self, conn).await?;
        Ok(list.pop())
    }
    @%- if !config.force_disable_cache %@

    pub async fn __select_for_cache(self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Cache>> {
        let result: Vec<(CacheData, _)> = self._select(conn).await?;
        let time = MSec::now();
        let list = result.into_iter().map(|(v, f)| (Arc::new(CacheWrapper::from_inner(v, conn.shard_id(), time)), f).into()).collect();
        Ok(list)
    }
    @%- endif %@
    @%- if def.use_cache() %@

    pub async fn select_from_cache(self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Cache>> {
        self._select_from_cache(conn).await
    }
    @%- endif %@

    pub async fn count(self, conn: &mut DbConn) -> Result<i64> {
        let result: Count = self.select_one_for(conn).await?.unwrap_or_default();
        Ok(result.c)
    }

    pub async fn select_stream(self, conn: &mut DbConn) -> Result<impl Stream<Item = _@{ pascal_name }@>> {
        let mut rx = self._select_stream(Data::_sql_cols(@{ is_mysql_str }@), conn).await?;
        Ok(async_stream::stream! {
            while let Some((v, flags)) = rx.recv().await {
                match Data::from_row(&v) {
                    Ok(v) => {
                        yield _@{ pascal_name }@::from((v, flags));
                    }
                    Err(e) => {
                        error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{}", e);
                        break;
                    }
                }
            }
        })
    }

    pub async fn select_stream_for<T>(self, conn: &mut DbConn) -> Result<impl Stream<Item = T>>
    where
        T: for<'r> sqlx::FromRow<'r, DbRow> + SqlColumns + Send + Sync + Unpin + 'static,
    {
        let mut rx = self._select_stream(T::_sql_cols(@{ is_mysql_str }@), conn).await?;
        Ok(async_stream::stream! {
            while let Some((v, flags)) = rx.recv().await {
                match T::from_row(&v) {
                    Ok(v) => {
                        yield v;
                    }
                    Err(e) => {
                        error!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{}", e);
                        break;
                    }
                }
            }
        })
    }
    @#- @%- if !def.disable_update() || def.soft_delete().is_some() %@ #@
    @%- if !def.disable_update() %@

    #[allow(unused_mut)]
    #[allow(clippy::if_same_then_else)]
    pub async fn update(self, conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<u64> {
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let force_indexes = make_force_indexes(&filter_digest);
        @%- if def.updated_at_conf().is_some() %@
        if obj._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
            obj.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
        }
        @%- endif %@
        let mut vec: Vec<String> = Vec::new();
        @{- def.non_primaries_except_read_only()|fmt_join_cache_or_not("
        crate::misc::assign_sql_no_cache_update!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, \"{placeholder}\");", "
        crate::misc::assign_sql_no_cache_update!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, \"{placeholder}\");", "") }@
        let mut sql = format!(
            r#"UPDATE {}@{ table_name|db_esc }@ as _t1 SET {} {} {} {}"#,
            if !force_indexes.is_empty() {
                format!("/*+ INDEX(_t1 {}) */ ", force_indexes.join(","))
            } else {
                String::new()
            },
            &vec.join(","),
            Filter_::write_where(
                &self.filter,
                self.trash_mode,
                TRASHED_SQL,
                NOT_TRASHED_SQL,
                ONLY_TRASHED_SQL,
                conn.shard_id(),
            ),
            &self.raw_query,
            Order_::write_order(&self.order, &self.raw_order)
        );
        if let Some(limit) = self.limit {
            write!(sql, " limit {}", limit)?;
        }
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        @{- def.non_primaries_except_read_only()|fmt_join("
        for _n in 0..obj._op.{ident}.get_bind_num({may_null}) {
            query = query.bind(obj._update.{ident}{bind_as});
        }","") }@
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "update_with_filter", filter = format!("{:?}", &self.filter), ctx = conn.ctx_no(); "{}", &obj);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
        if let Some(c) = self.filter {
            query = c.bind_to_query(query);
        }
        for value in self.bind.into_iter() {
            query = value.bind(query);
        }
        let now = std::time::Instant::now();
        let result = if conn.wo_tx() {
            query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        } else {
            query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        };
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        @%- if !config.force_disable_cache %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            conn.push_cache_op(CacheOp::InvalidateAll.wrap()).await;
        }
        @%- endif %@
        Ok(result.rows_affected())
    }
    @%- endif %@
    @%- if !def.disable_delete() %@

    #[allow(unused_mut)]
    pub async fn delete(self, conn: &mut DbConn) -> Result<u64> {
        @{- def.soft_delete_tpl2("
        self.force_delete(conn).await","
        let mut obj = _repo_::updater();
        obj.mut_deleted_at().set(Some({val}.into()));
        self.update(conn, obj).await","
        let mut obj = _repo_::updater();
        obj.mut_deleted().set(true);
        self.update(conn, obj).await","
        let mut obj = _repo_::updater();
        let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});
        obj.mut_deleted().set(deleted);
        self.update(conn, obj).await")}@
    }

    #[allow(unused_mut)]
    #[allow(clippy::if_same_then_else)]
    pub async fn force_delete(self, conn: &mut DbConn) -> Result<u64> {
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let force_indexes = make_force_indexes(&filter_digest);
        let mut sql = format!(
            r#"DELETE {}FROM @{ table_name|db_esc }@ as _t1 {} {} {}"#,
            if !force_indexes.is_empty() {
                format!("/*+ INDEX(_t1 {}) */ ", force_indexes.join(","))
            } else {
                String::new()
            },
            Filter_::write_where(
                &self.filter,
                self.trash_mode,
                TRASHED_SQL,
                NOT_TRASHED_SQL,
                ONLY_TRASHED_SQL,
                conn.shard_id(),
            ),
            &self.raw_query,
            Order_::write_order(&self.order, &self.raw_order)
        );
        if let Some(limit) = self.limit {
            write!(sql, " limit {}", limit)?;
        }
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "delete_with_filter", filter = format!("{:?}", &self.filter), ctx = conn.ctx_no(); "");
        if let Some(c) = self.filter {
            query = c.bind_to_query(query);
        }
        for value in self.bind.into_iter() {
            query = value.bind(query);
        }
        let now = std::time::Instant::now();
        let result = if conn.wo_tx() {
            query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        } else {
            query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        };
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        @%- if !config.force_disable_cache %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            conn.push_cache_op(CacheOp::InvalidateAll.wrap()).await;
        }
        @%- endif %@
        Ok(result.rows_affected())
    }
    @%- else %@
    pub(crate) async fn delete(self, _conn: &mut DbConn) -> Result<u64> {
        Ok(0)
    }
    @%- endif %@
}

#[inline(never)]
fn write_filter_flag(cols: &str, filter_flag: &BTreeMap<&'static str, Filter_>, shard_id: ShardId) -> String {
    let mut buf = String::with_capacity(1000);
    buf.push_str(cols);
    for (name, filter) in filter_flag {
        let mut trash_mode: TrashMode = Default::default();
        buf.push_str(", (");
        filter.write(&mut buf, 1, &mut trash_mode, shard_id, false);
        buf.truncate(buf.len() - " AND ".len());
        buf.push_str(") as ");
        buf.push_str(name);
    }
    buf
}

#[async_trait]
pub trait UnionBuilder {
    async fn select_union(self, conn: &mut DbConn, order: Option<Vec<Order_>>, limit: Option<usize>, offset: Option<usize>) -> Result<Vec<_@{ pascal_name }@>>;
    async fn select_union_for_update(self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Updater>>;
}
@%- if !config.force_disable_cache %@

#[async_trait]
pub(crate) trait _UnionBuilder {
    async fn __select_union_for_cache(self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Cache>>;
}
@%- endif %@

async fn _union(
    mut list: Vec<QueryBuilder>,
    conn: &mut DbConn,
    sql_cols: &str,
    order: Option<Vec<Order_>>,
    limit: Option<usize>,
    offset: Option<usize>,
    for_update: bool,
) -> Result<Vec<DbRow>>
{
    if order.is_none() && limit.is_none() && offset.is_none() {
        let mut result = Vec::new();
        loop {
            let chunk: Vec<_> = list.drain(list.len().saturating_sub(db::UNION_LIMIT)..).collect();
            if chunk.is_empty() {
                break;
            }
            let mut sql = chunk
                .iter()
                .map(|v| {
                    let filter_digest = v.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
                    format!("({})", v._sql(sql_cols, for_update, conn.shard_id(), &filter_digest))
                })
                .collect::<Vec<_>>()
                .join(" UNION ALL ");
            @%- if !config.is_mysql() %@
            let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
            @%- endif %@
            let mut query = sqlx::query(&sql);
            debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
            for builder in chunk {
                query = builder._bind(query, true);
            }
            result.append(&mut crate::misc::fetch!(conn, query, fetch_all));
        }
        Ok(result)
    } else {
        let mut sql = list
            .iter()
            .map(|v| {
                let filter_digest = v.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
                format!("({})", v._sql(sql_cols, for_update, conn.shard_id(), &filter_digest))
            })
            .collect::<Vec<_>>()
            .join(" UNION ");
        write!(sql, " {}", Order_::write_order(&order, &None))?;
        if let Some(limit) = limit {
            write!(sql, " limit {}", limit)?;
        }
        if let Some(offset) = offset {
            write!(sql, " offset {}", offset)?;
        }
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        for builder in list {
            query = builder._bind(query, true);
        }
        Ok(crate::misc::fetch!(conn, query, fetch_all))
    }
}

#[async_trait]
impl UnionBuilder for Vec<QueryBuilder> {
    async fn select_union(mut self, conn: &mut DbConn, order: Option<Vec<Order_>>, limit: Option<usize>, offset: Option<usize>) -> Result<Vec<_@{ pascal_name }@>> {
        if self.is_empty() {
            return Ok(Vec::new());
        }
        let result = _union(self, conn, Data::_sql_cols(@{ is_mysql_str }@), order, limit, offset, false).await?;
        let result: sqlx::Result<Vec<_>> = result.iter().map(Data::from_row).collect();
        Ok(result?.into_iter().map(|v| (v, BTreeMap::default()).into()).collect())
    }
    async fn select_union_for_update(mut self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Updater>> {
        if self.is_empty() {
            return Ok(Vec::new());
        }
        let result = _union(self, conn, Data::_sql_cols(@{ is_mysql_str }@), None, None, None, true).await?;
        let result: sqlx::Result<Vec<_>> = result.iter().map(Data::from_row).collect();
        Ok(result?.into_iter().map(|v| (v, BTreeMap::default()).into()).collect())
    }
}
@%- if !config.force_disable_cache %@

#[async_trait]
impl _UnionBuilder for Vec<QueryBuilder> {
    async fn __select_union_for_cache(self, conn: &mut DbConn) -> Result<Vec<_@{ pascal_name }@Cache>> {
        let result = _union(self, conn, CacheData::_sql_cols(@{ is_mysql_str }@), None, None, None, false).await?;
        let result: sqlx::Result<Vec<_>> = result.iter().map(CacheData::from_row).collect();
        let time = MSec::now();
        let list = result?.into_iter().map(|v| (Arc::new(CacheWrapper::from_inner(v, conn.shard_id(), time)), BTreeMap::default()).into()).collect();
        Ok(list)
    }
}
@%- endif %@

fn id_to_string(id: &InnerPrimary) -> String {
    format!("@{ def.primaries()|fmt_join("{col}={disp}", ", ") }@"@{ def.primaries()|fmt_join(", id.{index}", "") }@)
}

#[allow(clippy::useless_format)]
fn primaries_to_str(s: &[InnerPrimary]) -> String {
    let v = s.iter().fold(Vec::new(), |mut i, p| {
        i.push(format!("@{ def.primaries()|fmt_join_with_paren("{disp}", ", ") }@"@{ def.primaries()|fmt_join(", p.{index}", "") }@));
        i
    });
    format!("@{ def.primaries()|fmt_join_with_paren("{col}", ", ") }@={}", v.join(","))
}

@%- if !config.force_disable_cache %@

pub struct _@{ pascal_name }@Cache_;

impl _@{ pascal_name }@Cache_ {
    #[allow(clippy::needless_if)]
    pub async fn invalidate_cache<T>(conn: &DbConn, id: T) -> Result<()>
    where
        T: Into<Primary>,
    {
        if USE_CACHE || ENABLE_ALL_ROWS_CACHE {
            @%- if def.act_as_job_queue() %@
            @%- else if def.clear_all_cache_on_update() %@
            let sync = DbConn::inc_cache_sync(conn.shard_id()).await;
            let mut sync_map = FxHashMap::default();
            sync_map.insert(conn.shard_id(), sync);
            CacheMsg(vec![crate::CacheOp::_AllClear], sync_map)
                .do_send()
                .await;
            @%- else %@
            let id: InnerPrimary = (&id.into()).into();
            let sync = DbConn::inc_cache_sync(conn.shard_id()).await;
            let mut sync_map = FxHashMap::default();
            sync_map.insert(conn.shard_id(), sync);
            CacheMsg(vec![CacheOp::Invalidate{id, shard_id: conn.shard_id()}.wrap()], sync_map)
                .do_send()
                .await;
            @%- endif %@
        }
        Ok(())
    }
    @%- if def.use_cache() %@
    pub async fn find_self(_self: &_@{ pascal_name }@Cache, conn: &mut DbConn, joiner: Option<Box<Joiner_>>) -> Result<_@{ pascal_name }@Cache> {
        @%- if def.is_soft_delete() %@
        _repo_::find_from_cache_with_trashed(conn, Primary::from(_self), joiner).await
        @%- else %@
        _repo_::find_from_cache(conn, Primary::from(_self), joiner).await
        @%- endif %@
    }
    @%- endif %@
}
@%- endif %@

pub mod _repo_ {
    use super::*;

    pub fn updater() -> _@{ pascal_name }@Updater {
        _@{ pascal_name }@Updater {
            _data: Data::default(),
            _update: Data::default(),
            _filter_flag: Default::default(),
            _is_new: false,
            _do_delete: false,
            _upsert: false,
            _is_loaded: false,
            _op: OpData::default(),
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("\n            {rel_name}: None,", "") }@
        }
    }

    pub fn updater_of<T>(id: T) -> _@{ pascal_name }@Updater
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        _@{ pascal_name }@Updater {
            _data: Data {
                @{- def.primaries()|fmt_join("
                {ident}: id.{index}{raw_to_inner},", "") }@
                ..Data::default()
            },
            _update: Data::default(),
            _filter_flag: Default::default(),
            _is_new: false,
            _do_delete: false,
            _upsert: false,
            _is_loaded: false,
            _op: OpData::default(),
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("\n            {rel_name}: None,", "") }@
        }
    }
@%- for parent in def.downcast_aggregation() %@

    pub fn downcast_from(base: &db::models::@{ parent.group_name|snake|ident }@::@{ parent.name|snake|ident }@::_@{ parent.name|pascal }@) -> Option<_@{ pascal_name }@> {
        if base._inner.@{ def.inheritance_check() }@ {
            let clone = base._inner.clone();
            Some(_@{ pascal_name }@ {
                _inner: Data {
                    @{- def.all_fields()|fmt_join("
                    {ident}: clone.{ident},", "") }@
                },
@{- def.relations_one_and_belonging(Joinable::Join, false)|fmt_rel_join("\n                {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n                {rel_name}: None,", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("\n                {rel_name}: None,", "") }@
            })
        } else {
            None
        }
    }
@%- endfor %@
@%- for parent in def.downcast_simple() %@

    pub fn downcast_from(base: &db::models::@{ parent.group_name|snake|ident }@::@{ parent.name|snake|ident }@::_@{ parent.name|pascal }@) -> _@{ pascal_name }@ {
        let clone = base._inner.clone();
        _@{ pascal_name }@ {
            _inner: Data {
                @{- def.all_fields()|fmt_join("
                {ident}: clone.{ident},", "") }@
            },
@{- def.relations_one_and_belonging(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n            {rel_name}: None,", "") }@
@{- def.relations_belonging_outer_db(Joinable::Join, false)|fmt_rel_outer_db_join("\n            {rel_name}: None,", "") }@
        }
    }
@%- endfor %@

@%- if def.enable_all_rows_cache() %@
@%- if def.enable_filtered_rows_cache() %@

    #[allow(clippy::if_same_then_else)]
    pub async fn find_all_from_cache(
        conn: &DbConn,
        filter: Option<Filter_>,
        filter_flag: BTreeMap<&'static str, Filter_>,
        order: Option<Vec<Order_>>,
        limit: Option<usize>,
    ) -> Result<Arc<Vec<_@{ pascal_name }@Cache>>> {
        static SEMAPHORE: Lazy<Vec<Semaphore>> = Lazy::new(|| DbConn::shard_num_range().map(|_| Semaphore::new(1)).collect());
        let shard_id = conn.shard_id();
        let ctx_no = conn.ctx_no();
        if let Some(arc) = ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].load_full() {
            return Ok(arc);
        }
        let _guard = SEMAPHORE[shard_id as usize].acquire().await?;
        if let Some(arc) = ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].load_full() {
            return Ok(arc);
        }
        let mut conn = DbConn::_new_with_ctx(ctx_no, shard_id);
        conn.begin_cache_tx().await?;
        let filter_digest = self.filter.as_ref().map(|f| f.to_string()).unwrap_or_default();
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "filter digest:{}", filter_digest);
        let force_indexes = make_force_indexes(&filter_digest);
        let mut sql = format!(
            r#"SELECT {}{} FROM @{ table_name|db_esc }@ as _t1 {} {}"#,
            if !force_indexes.is_empty() {
                format!("/*+ INDEX(_t1 {}) */ ", force_indexes.join(","))
            } else {
                String::new()
            },
            CacheData::_sql_cols(@{ is_mysql_str }@),
            Filter_::write_where(
                &filter,
                TrashMode::Not,
                TRASHED_SQL,
                NOT_TRASHED_SQL,
                ONLY_TRASHED_SQL,
                conn.shard_id(),
            ),
            Order_::write_order(&order, &None)
        );
        if let Some(limit) = limit {
            write!(sql, " limit {}", limit)?;
        }
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        if let Some(c) = filter {
            query = c.bind_to_query(query);
        }
        let now = std::time::Instant::now();
        let result = crate::misc::fetch!(conn, query, fetch_all);
        if now.elapsed() > std::time::Duration::from_secs(1) {
            warn!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(); "[SLOW QUERY] time={}s digest={:?}", now.elapsed().as_millis() as f64 / 1000.0, filter_digest);
        }
        let result: sqlx::Result<Vec<_>> = result.iter().map(CacheData::from_row).collect();
        let time = MSec::now();
        let mut list: Vec<CacheWrapper> = result?.into_iter().map(|data| CacheWrapper::from_inner(data, shard_id, time)).collect();
        @{- def.relations_in_cache(Joinable::Join)|fmt_rel_join("\n        CacheWrapper::fetch_{raw_rel_name}_for_vec(&mut list, &mut conn).await?;", "") }@
        let list: Vec<_@{ pascal_name }@Cache> = list.into_iter().map(|v| Arc::new(v).into()).collect();
        let arc = Arc::new(list);
        let sync = CACHE_RESET_SYNC_ALL_ROWS.get().unwrap()[shard_id as usize].lock().await;
        if *sync <= conn.cache_sync() {
            ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].swap(Some(Arc::clone(&arc)));
        }
        Ok(arc)
    }
@%- else %@

    pub async fn find_all_from_cache(
        conn: &DbConn,
        order: Option<Vec<Order_>>,
    ) -> Result<Arc<Vec<_@{ pascal_name }@Cache>>> {
        static SEMAPHORE: Lazy<Vec<Semaphore>> = Lazy::new(|| DbConn::shard_num_range().map(|_| Semaphore::new(1)).collect());
        let shard_id = conn.shard_id();
        let ctx_no = conn.ctx_no();
        if let Some(arc) = ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].load_full() {
            return Ok(arc);
        }
        let _guard = SEMAPHORE[shard_id as usize].acquire().await?;
        if let Some(arc) = ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].load_full() {
            return Ok(arc);
        }
        let mut conn = DbConn::_new_with_ctx(ctx_no, shard_id);
        conn.begin_cache_tx().await?;
        let mut sql = format!(
            r#"SELECT {} FROM @{ table_name|db_esc }@ as _t1 {} {}"#,
            CacheData::_sql_cols(@{ is_mysql_str }@),
            Filter_::write_where(
                &None,
                TrashMode::Not,
                TRASHED_SQL,
                NOT_TRASHED_SQL,
                ONLY_TRASHED_SQL,
                conn.shard_id(),
            ),
            Order_::write_order(&order, &None)
        );
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        let result = crate::misc::fetch!(conn, query, fetch_all);
        let result: sqlx::Result<Vec<_>> = result.iter().map(CacheData::from_row).collect();
        let time = MSec::now();
        #[allow(clippy::needless_collect)]
        let mut list: Vec<CacheWrapper> = result?.into_iter().map(|data| CacheWrapper::from_inner(data, shard_id, time)).collect();
        @{- def.relations_in_cache(Joinable::Join)|fmt_rel_join("\n        CacheWrapper::fetch_{raw_rel_name}_for_vec(&mut list, &mut conn).await?;", "") }@
        let list: Vec<_@{ pascal_name }@Cache> = list.into_iter().map(|v| Arc::new(v).into()).collect();
        let arc = Arc::new(list);
        let sync = CACHE_RESET_SYNC_ALL_ROWS.get().unwrap()[shard_id as usize].lock().await;
        if *sync <= conn.cache_sync() {
            ALL_ROWS_CACHE.get().unwrap()[shard_id as usize].swap(Some(Arc::clone(&arc)));
        }
        Ok(arc)
    }
@%- endif %@
@%- endif %@

    pub fn query() -> QueryBuilder {
        QueryBuilder::default()
    }
    @%- if !config.force_disable_cache %@

    pub async fn clear_cache() -> Result<()> {
        @%- if def.act_as_job_queue() %@
        @%- else if def.clear_all_cache_on_update() %@
        let sync_map = DbConn::inc_all_cache_sync().await;
        CacheMsg(vec![crate::CacheOp::_AllClear], sync_map)
            .do_send()
            .await;
        @%- else %@
        let sync_map = DbConn::inc_all_cache_sync().await;
        CacheMsg(vec![CacheOp::InvalidateAll.wrap()], sync_map)
            .do_send()
            .await;
        @%- endif %@
        Ok(())
    }
    @%- endif %@

    pub async fn find<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<_@{ pascal_name }@>
    where
        T: Into<Primary>,
    {
        let id: Primary = id.into();
        find_optional(conn, id.clone(), joiner, filter, filter_flag)
            .await?
            .with_context(|| err::RowNotFound::new(TABLE_NAME, id_to_string(&(&id).into())))
    }
    @%- if def.is_soft_delete() %@

    pub async fn find_with_trashed<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<_@{ pascal_name }@>
    where
        T: Into<Primary>,
    {
        let id: Primary = id.into();
        find_optional_with_trashed(conn, id.clone(), joiner, filter, filter_flag)
            .await?
            .with_context(|| err::RowNotFound::new(TABLE_NAME, id_to_string(&(&id).into())))
    }
    @%- endif %@
    @%- if def.use_cache() %@

    pub async fn find_from_cache<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>) -> Result<_@{ pascal_name }@Cache>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        __find_optional_from_cache(conn, id.clone(), joiner)
            .await?
@{- def.soft_delete_tpl("","
            .filter(|data| data._wrapper._inner.deleted_at.is_none())","
            .filter(|data| data._wrapper._inner.deleted == 0)")}@
            .with_context(|| err::RowNotFound::new(TABLE_NAME, id_to_string(&id)))
    }
    @%- if def.is_soft_delete() %@

    pub async fn find_from_cache_with_trashed<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>) -> Result<_@{ pascal_name }@Cache>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        __find_optional_from_cache(conn, id.clone(), joiner)
            .await?
            .with_context(|| err::RowNotFound::new(TABLE_NAME, id_to_string(&id)))
    }
    @%- endif %@
    @%- endif %@

    pub fn list_to_map(list: Vec<_@{ pascal_name }@>) -> AHashMap<Primary, _@{ pascal_name }@> {
        let mut map = AHashMap::default();
        for v in list {
            map.insert((&v).into(), v);
        }
        map
    }
    @%- if def.use_cache() %@

    pub fn cache_list_to_map(list: Vec<_@{ pascal_name }@Cache>) -> AHashMap<Primary, _@{ pascal_name }@Cache> {
        let mut map = AHashMap::default();
        for v in list {
            map.insert((&v).into(), v);
        }
        map
    }
    @%- endif %@

    pub fn updater_list_to_map(list: Vec<_@{ pascal_name }@Updater>) -> AHashMap<Primary, _@{ pascal_name }@Updater> {
        let mut map = AHashMap::default();
        for v in list {
            map.insert((&v).into(), v);
        }
        map
    }

    pub async fn find_many<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Vec<_@{ pascal_name }@>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        let mut list = __find_many(conn, &ids, TrashMode::Not, filter, filter_flag.unwrap_or_default()).await?.into_iter().map(|v| v.into()).collect();
        _@{ pascal_name }@Joiner::join(&mut list, conn, joiner).await?;
        Ok(list)
    }
    @%- if def.is_soft_delete() %@

    pub async fn find_many_with_trashed<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Vec<_@{ pascal_name }@>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        let mut list = __find_many(conn, &ids, TrashMode::With, filter, filter_flag.unwrap_or_default()).await?.into_iter().map(|v| v.into()).collect();
        _@{ pascal_name }@Joiner::join(&mut list, conn, joiner).await?;
        Ok(list)
    }
    @%- endif %@
    @%- if def.use_cache() %@

    pub async fn find_many_from_cache<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>) -> Result<Vec<_@{ pascal_name }@Cache>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        Ok(__find_many_from_cache(conn, ids, joiner).await?.into_iter()@{ def.soft_delete_tpl("",".filter(|data| data._wrapper._inner.deleted_at.is_none())",".filter(|data| data._wrapper._inner.deleted == 0)")}@.collect())
    }
    @%- if def.is_soft_delete() %@

    pub async fn find_many_from_cache_with_trashed<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>) -> Result<Vec<_@{ pascal_name }@Cache>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        __find_many_from_cache(conn, ids, joiner).await
    }
    @%- endif %@
    @%- endif %@

    pub async fn find_optional<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Option<_@{ pascal_name }@>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        @%- if def.dummy_always_joinable() %@
        let data: Option<(Data, _)> = Some((Data {
            @{- def.primaries()|fmt_join("
            {ident}: id.{index}{raw_to_inner},", "") }@
            ..Default::default()
        }, Default::default()));
        @%- else %@
        let data: Option<_> = __find_optional(conn, Data::_sql_cols(@{ is_mysql_str }@), id, TrashMode::Not, filter, filter_flag.unwrap_or_default()).await?.map(|(v, f)| Data::from_row(&v).map(|v| (v, f))).transpose()?;
        @%- endif %@
        let mut obj = data.map(_@{ pascal_name }@::from);
        if let Some(obj) = obj.as_mut() {
            _@{ pascal_name }@Joiner::join(obj, conn, joiner).await?;
        }
        Ok(obj)
    }

    pub async fn exists<T>(conn: &mut DbConn, id: T, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Option<BTreeMap<&'static str, bool>>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        @%- if def.dummy_always_joinable() %@
        Ok(Default::default())
        @%- else %@
        let data: Option<_> = __find_optional(conn, Exists::_sql_cols(@{ is_mysql_str }@), id, TrashMode::Not, filter, filter_flag.unwrap_or_default()).await?.map(|(_, f)| f);
        Ok(data)
        @%- endif %@
    }
    @%- if !def.disable_update() %@

    pub async fn find_optional_for_update<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Option<_@{ pascal_name }@Updater>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        let result = __find_for_update(conn, &id, TrashMode::Not, filter, filter_flag.unwrap_or_default()).await?;
        let mut obj = result.map(__Updater__::from);
        if let Some(obj) = obj.as_mut() {
            _@{ pascal_name }@Joiner::join(obj, conn, joiner).await?;
        }
        Ok(obj)
    }
    @%- endif %@
    @%- if def.is_soft_delete() %@

    pub async fn find_optional_with_trashed<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Option<_@{ pascal_name }@>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        @%- if def.dummy_always_joinable() %@
        let data: Option<Data> = Some(Data {
            @{- def.primaries()|fmt_join("
            {ident}: id.{index}{raw_to_inner},", "") }@
            ..Default::default()
        });
        @%- else %@
        let data: Option<_> = __find_optional(conn, Data::_sql_cols(@{ is_mysql_str }@), id, TrashMode::With, filter, filter_flag.unwrap_or_default()).await?.map(|(v, f)| Data::from_row(&v).map(|v| (v, f))).transpose()?;
        @%- endif %@
        let mut obj = data.map(_@{ pascal_name }@::from);
        if let Some(obj) = obj.as_mut() {
            _@{ pascal_name }@Joiner::join(obj, conn, joiner).await?;
        }
        Ok(obj)
    }

    pub async fn exists_with_trashed<T>(conn: &mut DbConn, id: T, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Option<BTreeMap<&'static str, bool>>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        @%- if def.dummy_always_joinable() %@
        Ok(Default::default())
        @%- else %@
        let data: Option<_> = __find_optional(conn, Exists::_sql_cols(@{ is_mysql_str }@), id, TrashMode::With, filter, filter_flag.unwrap_or_default()).await?.map(|(_, f)| f);
        Ok(data)
        @%- endif %@
    }
    @%- endif %@
    @%- if def.use_cache() %@

    #[allow(clippy::needless_question_mark)]
    pub async fn find_optional_from_cache<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>) -> Result<Option<_@{ pascal_name }@Cache>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        Ok(__find_optional_from_cache(conn, id, joiner).await?@{- def.soft_delete_tpl("",".filter(|data| data._wrapper._inner.deleted_at.is_none())",".filter(|data| data._wrapper._inner.deleted == 0)")}@)
    }
    @%- if def.is_soft_delete() %@

    pub async fn find_optional_from_cache_with_trashed<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>) -> Result<Option<_@{ pascal_name }@Cache>>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        __find_optional_from_cache(conn, id, joiner).await
    }
    @%- endif %@
    @%- endif %@
    @%- if !def.disable_update() %@

    pub async fn find_for_update<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<_@{ pascal_name }@Updater>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        let result = __find_for_update(conn, &id, TrashMode::Not, filter, filter_flag.unwrap_or_default()).await?@{ def.soft_delete_tpl("",".filter(|(data, _)| data.deleted_at.is_none())",".filter(|(data, _)| data.deleted == 0)")}@;
        let data = result.with_context(|| err::RowNotFound::new(TABLE_NAME, id.to_string()))?;
        let mut obj = __Updater__::from(data);
        _@{ pascal_name }@Joiner::join(&mut obj, conn, joiner).await?;
        Ok(obj)
    }
    @%- endif %@
    @%- if !def.disable_update() && def.is_soft_delete() %@

    pub async fn find_for_update_with_trashed<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<_@{ pascal_name }@Updater>
    where
        T: Into<Primary>,
    {
        let id: InnerPrimary = (&id.into()).into();
        let result = __find_for_update(conn, &id, TrashMode::With, filter, filter_flag.unwrap_or_default()).await?;
        let data = result.with_context(|| err::RowNotFound::new(TABLE_NAME, id.to_string()))?;
        let mut obj = __Updater__::from(data);
        _@{ pascal_name }@Joiner::join(&mut obj, conn, joiner).await?;
        Ok(obj)
    }
    @%- endif %@
@%- if !def.disable_update() %@

    pub async fn find_many_for_update<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Vec<_@{ pascal_name }@Updater>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        __find_many_for_update(conn, &ids, TrashMode::Not, joiner, filter, filter_flag.unwrap_or_default()).await
    }

    pub async fn find_many_for_update_with_trashed<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: Option<BTreeMap<&'static str, Filter_>>) -> Result<Vec<_@{ pascal_name }@Updater>>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        __find_many_for_update(conn, &ids, TrashMode::With, joiner, filter, filter_flag.unwrap_or_default()).await
    }
@%- endif %@
@%- for (index_name, index) in def.unique_index() %@
    pub async fn find_by_@{ index_name }@<@{ index.fields(index_name, def)|fmt_index_col("T{index}", ", ") }@>(conn: &mut DbConn, @{ index.fields(index_name, def)|fmt_index_col("_{name}: T{index}", ", ") }@, joiner: Option<Box<Joiner_>>) -> Result<_@{ pascal_name }@>
    where
    @{- index.fields(index_name, def)|fmt_index_col("
        T{index}: Into<{filter_type}>,", "") }@
    {
        @{- index.fields(index_name, def)|fmt_index_col("
        let val{index}: {filter_type} = _{name}.into();", "") }@
        let filter = Filter_::And(vec![@{- index.fields(index_name, def)|fmt_index_col("Filter_::EqKey(ColKey_::{ident}(val{index}.clone().into()))", ", ") }@]);
        query().filter(filter).join(joiner).select(conn).await?.pop()
            .with_context(|| err::RowNotFound::new(TABLE_NAME, format!("@{ index.fields(index_name, def)|fmt_index_col("{col_name}={}", ", ") }@", @{ index.fields(index_name, def)|fmt_index_col("val{index}", ", ") }@)))
    }
@%- endfor %@
@%- if def.use_cache() %@
@%- for (index_name, index) in def.unique_index() %@
    pub async fn find_by_@{ index_name }@_from_cache<@{ index.fields(index_name, def)|fmt_index_col("T{index}", ", ") }@>(conn: &mut DbConn, @{ index.fields(index_name, def)|fmt_index_col("_{name}: T{index}", ", ") }@, joiner: Option<Box<Joiner_>>) -> Result<_@{ pascal_name }@Cache>
    where
    @{- index.fields(index_name, def)|fmt_index_col("
        T{index}: Into<{filter_type}>,", "") }@
    {
        @{- index.fields(index_name, def)|fmt_index_col("
        let c{index}: {filter_type} = _{name}.into();", "") }@
        let key = VecColKey(vec![@{- index.fields(index_name, def)|fmt_index_col("ColKey_::{ident}(c{index}.clone().into())", ", ") }@]);
        if let Some(id) = Cache::get::<PrimaryWrapper>(&key, conn.shard_id(), true).await {
            if let Some(obj) = find_optional_from_cache(conn, &id.0, joiner.clone()).await? {
                if @{ index.fields(index_name, def)|fmt_index_col("obj._{raw_name}(){filter_check_eq}", " && ") }@ {
                    return Ok(obj);
                }
            }
        }
        let filter = Filter_::And(vec![@{- index.fields(index_name, def)|fmt_index_col("Filter_::EqKey(ColKey_::{ident}(c{index}.clone().into()))", ", ") }@]);
        let mut conn = DbConn::_new_with_ctx(conn.ctx_no(), conn.shard_id());
        conn.begin_cache_tx().await?;
        let obj = query().filter(filter).join(joiner).select_from_cache(&mut conn).await?.pop()
            .with_context(|| err::RowNotFound::new(TABLE_NAME, format!("@{ index.fields(index_name, def)|fmt_index_col("{col_name}={}", ", ") }@", @{ index.fields(index_name, def)|fmt_index_col("c{index}", ", ") }@)))?;
        let id = PrimaryWrapper(InnerPrimary::from(&obj), conn.shard_id(), MSec::now());
        Cache::insert_long(&key, Arc::new(id), true).await;
        Ok(obj)
    }
@%- endfor %@
@%- if !def.act_as_job_queue() && !def.clear_all_cache_on_update() %@

    pub async fn insert_dummy_cache(conn: &DbConn, obj: _@{ pascal_name }@Updater) -> Result<()> {
        let _lock = db::models::CACHE_UPDATE_LOCK.write().await;
        let cache_msg = CacheOp::Insert {
            overwrite: false,
            shard_id: conn.shard_id(),
            data: obj._data,
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: None,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: None,", "") }@
        };
        let mut sync_map = FxHashMap::default();
        sync_map.insert(conn.shard_id(), 0);
        cache_msg.handle_cache_msg(Arc::new(sync_map)).await;
        Ok(())
    }
@%- endif %@
@%- endif %@

    pub async fn save(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<Option<_@{ pascal_name }@>> {
        obj.__validate()?;
        obj.__set_default_value(conn).await?;
        __save(conn, obj, 0).await
    }

    pub async fn overwrite(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<Option<_@{ pascal_name }@>> {
        obj.__validate()?;
        ensure!(obj.is_new(), "The obj is not new.");
        obj.__set_default_value(conn).await?;
        __save(conn, obj, 1).await
    }
    @%- if !def.disable_update() %@

    /// update_many ignores versions.
    pub async fn update_many<I, T>(conn: &mut DbConn, ids: I, mut updater: _@{ pascal_name }@Updater) -> Result<u64>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        @%- if def.updated_at_conf().is_some() %@
        if updater._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
            updater.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
        }
        @%- endif %@
        __update_many(conn, ids, updater).await
    }
    @%- endif %@

    /// insert_ignore does not return an error if the error is negligible.
    /// insert_ignore does not save the related tables.
    #[allow(clippy::unnecessary_cast)]
    pub async fn insert_ignore(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<Option<_@{ pascal_name }@Updater>> {
        obj.__validate()?;
        ensure!(obj.is_new(), "The obj is not new.");
        obj.__set_default_value(conn).await?;
        @%- if config.is_mysql() %@
        let sql = r#"INSERT IGNORE INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) 
            VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@)"#;
        @%- else %@
        let sql = r#"INSERT INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) 
            VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@) ON CONFLICT DO NOTHING@{ def.auto_inc()|fmt_join(" RETURNING xmax,{col_esc}", "") }@;"#;
        let sql = &senax_common::convert_mysql_placeholders_to_postgresql(sql);
        @%- endif %@
        let query = bind_to_query(sqlx::query(sql), &obj._data);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        let (rows_affected, _last_insert_id) = conn.execute@{ def.auto_inc()|fmt_join("_with_last_insert_id", "") }@(query).await.context(err::ErrorTable(TABLE_NAME))?;
        if rows_affected == 0 {
            return Ok(None);
        }
@{- def.auto_inc()|fmt_join("
        if obj._data.{ident} == 0 {
            obj._data.{ident} = _last_insert_id as {inner};
        }", "") }@
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "insert_ignore", ctx = conn.ctx_no(); "{}", &obj);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
        obj._is_new = false;
        obj._op = OpData::default();
        @%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        conn.push_cache_op(CacheOp::Queued.wrap()).await;
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            let cache_msg = CacheOp::Insert {
                overwrite: false,
                shard_id: conn.shard_id(),
                data: obj._data.clone(),
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
            };
            conn.push_cache_op(cache_msg.wrap()).await;
        }
        @%- endif %@
        @%- endif %@
        Ok(Some(obj))
    }
    @%- if def.enable_delayed_insert() %@

    /// If delayed_insert is used, the data will be collectively registered later.
    pub async fn delayed_insert(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<()> {
        ensure!(obj.is_new(), "The obj is not new.");
        obj.__validate()?;
        obj.__set_default_value(conn).await?;
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "delayed_insert", ctx = conn.ctx_no(); "{}", &obj);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
        conn.push_callback(Box::new(|| {
            async move {
                DELAYED_INSERT_QUEUE.push(obj.into());
                if !db::is_test_mode() {
                    DelayedActor::handle(DelayedMsg::InsertFromMemory);
                } else {
                    handle_delayed_msg_insert_from_memory().await;
                }
            }.boxed()
        })).await;
        Ok(())
    }
    @%- endif %@
    @%- if def.enable_delayed_save() %@

    // The data will be updated collectively later.
    // delayed_save does not support relational tables.
    pub async fn delayed_save(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<()> {
        obj.__validate()?;
        obj.__set_default_value(conn).await?;
        if obj.will_be_deleted() {
            obj._op = OpData::default();
        }
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
        let shard_id = conn.shard_id() as usize;
        conn.push_callback(Box::new(move || {
            async move {
                DELAYED_SAVE_QUEUE.get().unwrap()[shard_id].push(obj);
                if !db::is_test_mode() {
                    DelayedActor::handle(DelayedMsg::Save);
                } else {
                    handle_delayed_msg_save().await;
                }
            }.boxed()
        })).await;
        Ok(())
    }
    @%- endif %@
    @%- if def.enable_delayed_update() %@

    // The data will be updated collectively later.
    // delayed_update does not support relational tables and version.
    pub async fn delayed_update(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<()> {
        if obj.is_new() {
            panic!("INSERT is not supported.");
        }
        @%- if def.soft_delete().is_none() %@
        if obj.will_be_deleted() {
            panic!("DELETE is not supported.");
        }
        @%- endif %@
        obj.__validate()?;
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
        let shard_id = conn.shard_id() as usize;
        conn.push_callback(Box::new(move || {
            async move {
                DELAYED_UPDATE_QUEUE.get().unwrap()[shard_id].push(obj);
                if !db::is_test_mode() {
                    DelayedActor::handle(DelayedMsg::Update);
                } else {
                    handle_delayed_msg_update().await;
                }
            }.boxed()
        })).await;
        Ok(())
    }
    @%- endif %@
    @%- if def.enable_delayed_upsert() %@

    // The data will be updated collectively later.
    // delayed_upsert does not support relational tables.
    pub async fn delayed_upsert(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<()> {
        obj.__validate()?;
        @%- if def.soft_delete().is_none() %@
        if obj.will_be_deleted() {
            panic!("DELETE is not supported.");
        }
        @%- endif %@
        obj.__set_default_value(conn).await?;
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
@{- def.relations_belonging(Joinable::Join, false)|fmt_rel_join("\n        obj.{rel_name} = None;", "") }@
        let shard_id = conn.shard_id() as usize;
        conn.push_callback(Box::new(move || {
            async move {
                DELAYED_UPSERT_QUEUE.get().unwrap()[shard_id].push(obj);
                if !db::is_test_mode() {
                    DelayedActor::handle(DelayedMsg::Upsert);
                } else {
                    handle_delayed_msg_upsert().await;
                }
            }.boxed()
        })).await;
        Ok(())
    }
    @%- endif %@

    /// If ignore is ture, bulk_insert does not guarantee that the related table will be saved correctly
    pub async fn bulk_insert(conn: &mut DbConn, mut list: Vec<_@{ pascal_name }@Updater>, ignore: bool) -> Result<()> {
        _bulk_insert(conn, list, ignore, false).await
    }
@%- if !def.disable_update() %@

    pub async fn bulk_replace(conn: &mut DbConn, mut list: Vec<_@{ pascal_name }@Updater>) -> Result<()> {
        _bulk_insert(conn, list, false, true).await
    }

    pub async fn bulk_overwrite(conn: &mut DbConn, mut list: Vec<_@{ pascal_name }@Updater>) -> Result<()> {
        let mut vec: Vec<ForInsert> = Vec::with_capacity(list.len());
        @%- if def.soft_delete().is_none() && !def.disable_delete() %@
        let mut remove_ids: Vec<Primary> = Vec::with_capacity(list.len());
        @%- endif %@
        for mut obj in list.into_iter() {
            obj.__validate()?;
            obj.__set_default_value(conn).await?;
            @%- if def.soft_delete().is_none() && !def.disable_delete() %@
            if obj.will_be_deleted() {
                remove_ids.push((&obj).into());
                continue;
            } else {
                obj.__set_overwrite_extra_value(conn);
            }
            @%- else %@
            obj.__set_overwrite_extra_value(conn);
            @%- endif %@
            vec.push(obj.into());
        }
        @%- if def.soft_delete().is_none() && !def.disable_delete() %@
        delete_by_ids(conn, remove_ids).await?;
        @%- endif %@
        __bulk_insert(conn, &vec, false, false, true).await
    }
@%- endif %@
    @%- if !def.disable_update() %@

    pub async fn bulk_upsert(conn: &mut DbConn, mut list: Vec<_@{ pascal_name }@Updater>, mut updater: _@{ pascal_name }@Updater) -> Result<()> {
        if list.is_empty() {
            return Ok(());
        }
        let mut vec = Vec::with_capacity(list.len());
        for mut obj in list.into_iter() {
            ensure!(obj.is_new(), "bulk_upsert supports only new objects.");
            obj.__validate()?;
            obj.__set_default_value(conn).await?;
            vec.push(obj._data);
        }
        @%- if def.updated_at_conf().is_some() %@
        if updater._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
            updater.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
        }
        @%- endif %@
        __bulk_upsert(conn, &vec, &updater).await
    }
    @%- endif %@
@%- if !def.disable_delete() %@

    #[allow(clippy::needless_borrow)]
    pub async fn delete_by_ids<I, T>(conn: &mut DbConn, ids: I) -> Result<u64>
    where
        I: IntoIterator<Item = T> + Send,
        T: Into<Primary>,
    {
@%- if def.soft_delete().is_some() %@
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        async fn inner(conn: &mut DbConn, ids: Vec<InnerPrimary>) -> Result<u64> {
            if ids.is_empty() {
                return Ok(0);
            }
            let mut rows_affected = 0u64;
            let id_chunks = ids.chunks(IN_CONDITION_LIMIT);
            @{- def.soft_delete_tpl2("","
            let deleted_at: {filter_type} = {val}.into();","","
            let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});")}@
            @%- if def.updated_at_conf().is_some() %@
            @%- let updated_at = def.get_updated_at() %@
            let updated_at: @{ updated_at.get_filter_type(false) }@ = @{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into();
            @%- endif %@
            for ids in id_chunks {
                let q = "@{ def.primaries()|fmt_join_with_paren("{placeholder}", ",") }@,".repeat(ids.len());
                let sql = format!(
                    r#"UPDATE @{ table_name|db_esc }@ SET @{ def.soft_delete_tpl2("","deleted_at=?","deleted=1","deleted=?")}@@% if def.updated_at_conf().is_some() %@, updated_at=?@%- endif %@ WHERE @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join_with_paren("{col_esc}", ",") }@ in ({});"#,
                    &q[0..q.len() - 1]
                );
                @%- if !config.is_mysql() %@
                let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
                @%- endif %@
                let mut query = sqlx::query(&sql);
                debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    @{- def.soft_delete_tpl2("","
                query = query.bind(deleted_at);","","
                query = query.bind(deleted);")}@
    @%- if def.updated_at_conf().is_some() %@
                query = query.bind(updated_at);
    @%- endif %@
                for id in ids {
                    @{- def.primaries()|fmt_join("
                    query = query.bind(id.{index}{bind_as});", "") }@
                }
                let result = if conn.wo_tx() {
                    query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
                } else {
                    query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
                };
                rows_affected += result.rows_affected();
            }
            debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "delete_by_ids", ctx = conn.ctx_no(), ids = primaries_to_str(&ids); "");
            @%- if !config.force_disable_cache %@
            @%- if def.act_as_job_queue() %@
            @%- else if def.clear_all_cache_on_update() %@
            conn.clear_all_cache = true;
            @%- else %@
            if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
                let mut updater = _repo_::updater();
                @{- def.soft_delete_tpl2("","
                updater.mut_deleted_at().set(Some(deleted_at));","
                updater.mut_deleted().set(true);","
                let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});
                updater.mut_deleted().set(deleted);")}@
    @%- if def.updated_at_conf().is_some() %@
                updater.mut_@{ ConfigDef::updated_at() }@().set(updated_at);
    @%- endif %@
                @{- def.non_primaries()|fmt_join_cache_or_not("", "
                updater._op.{ident} = Op::None;
                updater._update.{ident} = Default::default();", "") }@
                let cache_msg = CacheOp::UpdateMany {
                    ids,
                    shard_id: conn.shard_id(),
                    update: updater._update,
                    data_list: Vec::new(),
                    op: updater._op,
                };
                conn.push_cache_op(cache_msg.wrap()).await;
                @{- def.cache_owners|fmt_cache_owners("
                conn.push_cache_op(db::models::{mod}::CacheOp::InvalidateAll.wrap()).await;") }@
            }
            @%- endif %@
            @%- endif %@
            Ok(rows_affected)
        }
        inner(conn, ids).await
@%- else %@
        force_delete_by_ids(conn, ids).await
@%- endif %@
    }

    #[allow(clippy::needless_borrow)]
    pub async fn force_delete_by_ids<I, T>(conn: &mut DbConn, ids: I) -> Result<u64>
    where
        I: IntoIterator<Item = T>,
        T: Into<Primary>,
    {
        let ids: Vec<InnerPrimary> = ids.into_iter().map(|id| (&id.into()).into()).collect();
        async fn inner(conn: &mut DbConn, ids: Vec<InnerPrimary>) -> Result<u64> {
            if ids.is_empty() {
                return Ok(0);
            }
            let mut rows_affected = 0u64;
            let id_chunks = ids.chunks(IN_CONDITION_LIMIT);
            for ids in id_chunks {
                let q = "@{ def.primaries()|fmt_join_with_paren("{placeholder}", ",") }@,".repeat(ids.len());
                let sql = format!(
                    r#"DELETE FROM @{ table_name|db_esc }@ WHERE @{ def.primaries()|fmt_join_with_paren("{col_esc}", ",") }@ in ({});"#,
                    &q[0..q.len() - 1]
                );
                @%- if !config.is_mysql() %@
                let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
                @%- endif %@
                let mut query = sqlx::query(&sql);
                debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
                for id in ids {
                    @{- def.primaries()|fmt_join("
                    query = query.bind(id.{index}{bind_as});", "") }@
                }
                let result = if conn.wo_tx() {
                    query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
                } else {
                    query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
                };
                rows_affected += result.rows_affected();
            }
            debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "force_delete_by_ids", ctx = conn.ctx_no(), ids = primaries_to_str(&ids); "");
            @%- if !config.force_disable_cache %@
            @%- if def.act_as_job_queue() %@
            @%- else if def.clear_all_cache_on_update() %@
            conn.clear_all_cache = true;
            @%- else %@
            if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
                let shard_id = conn.shard_id();
                conn.push_cache_op(CacheOp::DeleteMany { ids, shard_id }.wrap()).await;
                @{- def.cache_owners|fmt_cache_owners("
                conn.push_cache_op(db::models::{mod}::CacheOp::InvalidateAll.wrap()).await;") }@
            }
            @%- endif %@
            @%- endif %@
            Ok(rows_affected)
        }
        inner(conn, ids).await
    }
@%- endif %@
@%- if !def.disable_delete() %@

    #[allow(unused_mut)]
    pub async fn delete(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<()> {
        @%- if def.updated_at_conf().is_some() %@
        if obj._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
            obj.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
        }
        @%- endif %@
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
        if !conn.clear_all_cache {
            @{- def.cache_owners|fmt_cache_owners("
            if let Some(v) = _base_filter_::repositories::{base_mod}::RelFk{rel_name_pascal}::get_fk(&obj._data) {
                if _base_filter_::repositories::{base_mod}::USE_CACHE || _base_filter_::repositories::{base_mod}::ENABLE_ALL_ROWS_CACHE || _base_filter_::repositories::{base_mod}::ENABLE_UPDATE_NOTICE {
                    let id: _base_filter_::repositories::{base_mod}::Primary = v.into();
                    let id: _base_filter_::repositories::{base_mod}::InnerPrimary = (&id).into();
                    conn.push_cache_op(_base_filter_::repositories::{base_mod}::CacheOp::Invalidate{id, shard_id: conn.shard_id()}.wrap()).await;
                }
            }") }@
        }
        @%- endif %@
        let cache_msg = __delete(conn, obj).await?;
        @%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            if let Some(cache_msg) = cache_msg {
                conn.push_cache_op(cache_msg.wrap()).await;
            }
        }
        @%- endif %@
        @%- endif %@
        Ok(())
    }
@%- endif %@
@%- if !def.disable_update() && def.soft_delete().is_some() %@

    #[allow(unused_mut)]
    pub async fn restore(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<_@{ pascal_name }@> {
        obj._do_delete = false;
        if !obj.has_been_deleted() {
            return Ok(obj.into());
        }
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
        if !conn.clear_all_cache {
            @{- def.cache_owners|fmt_cache_owners("
            _invalidate_{raw_rel_name}(conn, &obj._data).await;") }@
        }
        @%- endif %@
        @%- if def.updated_at_conf().is_some() %@
        if obj._op.@{ ConfigDef::updated_at()|ident }@ == Op::None {
            obj.mut_@{ ConfigDef::updated_at() }@().set(@{(def.updated_at_conf().unwrap() == Timestampable::RealTime)|if_then_else_ref("SystemTime::now()","conn.time()")}@.into());
        }
        @%- endif %@
@{- def.soft_delete_tpl2("","
        obj.mut_deleted_at().set(None);","
        obj.mut_deleted().set(false);","
        obj.mut_deleted().set(0);")}@
        let (obj, cache_msg) = __save_update(conn, obj).await?;
        @%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        conn.push_cache_op(CacheOp::Queued.wrap()).await;
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            if let Some(cache_msg) = cache_msg {
                conn.push_cache_op(cache_msg.wrap()).await;
            }
        }
        @%- endif %@
        @%- endif %@
        Ok(obj)
    }
@%- endif %@
@%- if !def.disable_delete() %@

    #[allow(clippy::needless_borrow)]
    pub async fn force_delete(conn: &mut DbConn, obj: _@{ pascal_name }@Updater) -> Result<()> {
        let id: InnerPrimary = (&obj).into();
        let sql = r#"DELETE FROM @{ table_name|db_esc }@ WHERE @{ def.primaries()|fmt_join("{col_esc}={placeholder}", " AND ") }@"#;
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        @{- def.primaries()|fmt_join("
        query = query.bind(id.{index}{bind_as});", "") }@
        if conn.wo_tx() {
            query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
        } else {
            query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
        }
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "force_delete", ctx = conn.ctx_no(), id = id.to_string(); "{}", &obj);
@%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            let shard_id = conn.shard_id();
            conn.push_cache_op(CacheOp::Delete { id, shard_id }.wrap()).await;
        }
        @%- endif %@
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
        if !conn.clear_all_cache {
            @{- def.cache_owners|fmt_cache_owners("
            _invalidate_{raw_rel_name}(conn, &obj._data).await;") }@
        }
        @%- endif %@
@%- endif %@
        Ok(())
    }
@%- endif %@

    pub async fn force_delete_all(conn: &mut DbConn) -> Result<()> {
        let query = sqlx::query(r#"DELETE FROM @{ table_name|db_esc }@"#);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        if conn.wo_tx() {
            query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
        } else {
            query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
        }
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "force_delete_all", ctx = conn.ctx_no(); "");
        @%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            conn.push_cache_op(CacheOp::DeleteAll {
                shard_id: conn.shard_id(),
            }.wrap()).await;
            @{- def.cache_owners|fmt_cache_owners("
            conn.push_cache_op(db::models::{mod}::CacheOp::InvalidateAll.wrap()).await;") }@
        }
        @%- endif %@
        @%- endif %@
        Ok(())
    }

    pub async fn truncate(conn: &mut DbConn) -> Result<()> {
        let query = sqlx::query(r#"TRUNCATE TABLE @{ table_name|db_esc }@"#);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "truncate", ctx = conn.ctx_no(); "");
        @%- if !config.force_disable_cache %@
        @%- if def.act_as_job_queue() %@
        @%- else if def.clear_all_cache_on_update() %@
        conn.clear_all_cache = true;
        @%- else %@
        if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
            conn.push_cache_op(CacheOp::DeleteAll {
                shard_id: conn.shard_id(),
            }.wrap()).await;
            @{- def.cache_owners|fmt_cache_owners("
            conn.push_cache_op(db::models::{mod}::CacheOp::InvalidateAll.wrap()).await;") }@
        }
        @%- endif %@
        @%- endif %@
        Ok(())
    }

    pub async fn select_virtual_row(conn: &mut DbConn, obj: &_@{ pascal_name }@Updater, filter_flag: BTreeMap<&'static str, Filter_>) -> Result<BTreeMap<&'static str, bool>>
    {
        let filter_flag_names: Vec<_> = filter_flag.keys().cloned().collect();
        let sql_cols = write_filter_flag("0", &filter_flag, conn.shard_id());
        let sql = format!(r#"SELECT {} FROM (SELECT @{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("? as {col_esc}", ", ") }@) as _t1"#, sql_cols);
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        for (_name, filter) in filter_flag {
            query = filter.bind_to_query(query);
        }
        let query = bind_to_query(query, &obj._data);
        let row = crate::misc::fetch!(conn, query, fetch_one);
        use sqlx::Row;
        let mut flags = BTreeMap::new();
        for name in filter_flag_names.clone() {
            flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
        }
        Ok(flags)
    }
}

async fn __find_many(
    conn: &mut DbConn,
    ids: &[InnerPrimary],
    trash_mode: TrashMode,
    filter: Option<Filter_>,
    filter_flag: BTreeMap<&'static str, Filter_>,
) -> Result<Vec<(Data, BTreeMap<&'static str, bool>)>> {
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let id_chunks = ids.chunks(IN_CONDITION_LIMIT);
    let mut list = Vec::with_capacity(ids.len());
    for ids in id_chunks {
        @%- if def.dummy_always_joinable() %@
        let mut v = ids
            .iter()
            .map(|id| {
                let id = id.clone();
                (Data {
                    @{- def.primaries()|fmt_join("
                    {ident}: id.{index}{raw_to_inner},", "") }@
                    ..Default::default()
                }, Default::default())
            })
            .collect();
        list.append(&mut v);
        @%- else %@
        let v = ___find_many(conn, Data::_sql_cols(@{ is_mysql_str }@), ids, trash_mode, filter.clone(), filter_flag.clone()).await?;
        let mut v: sqlx::Result<Vec<_>> = v.into_iter().map(|(v, f)| Data::from_row(&v).map(|v| (v, f))).collect();
        list.append(&mut v?);
        @%- endif %@
    }
    Ok(list)
}
#[allow(clippy::needless_borrow)]
async fn ___find_many(conn: &mut DbConn, sql_cols: &str, ids: &[InnerPrimary], trash_mode: TrashMode, filter: Option<Filter_>, filter_flag: BTreeMap<&'static str, Filter_>) -> Result<Vec<(DbRow, BTreeMap<&'static str, bool>)>>
{
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let filter_flag_names: Vec<_> = filter_flag.keys().cloned().collect();
    let sql_cols = write_filter_flag(sql_cols, &filter_flag, conn.shard_id());
    let mut filter_str = Filter_::write_where(&filter, trash_mode, TRASHED_SQL, NOT_TRASHED_SQL, ONLY_TRASHED_SQL, conn.shard_id());
    if filter_str.is_empty() {
        filter_str = "WHERE".to_string();
    } else {
        filter_str.push_str(" AND ");
    }
    let q = "@{ def.primaries()|fmt_join_with_paren("{placeholder}", ",") }@,".repeat(ids.len());
    let sql = format!(
        r#"SELECT {} FROM @{ table_name|db_esc }@ {filter_str} @{ def.primaries()|fmt_join_with_paren("{col_esc}", ",") }@ in ({});"#,
        sql_cols,
        &q[0..q.len() - 1]
    );
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let mut query = sqlx::query(&sql);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    for (_name, filter) in filter_flag {
        query = filter.bind_to_query(query);
    }
    if let Some(c) = filter {
        query = c.bind_to_query(query);
    }
    for id in ids {
        @{- def.primaries()|fmt_join("
        query = query.bind(id.{index}{bind_as});", "") }@
    }
    Ok(crate::misc::fetch!(conn, query, fetch_all)
        .into_iter()
        .map(|row| {
            use sqlx::Row;
            let mut flags = BTreeMap::new();
            for name in filter_flag_names.clone() {
                flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
            }
            (row, flags)
        })
        .collect())
}
@%- if def.use_cache() %@

async fn ___find_many_for_cache(
    conn: &mut DbConn,
    ids: &[InnerPrimary],
) -> Result<Vec<CacheWrapper>> {
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let id_chunks = ids.chunks(IN_CONDITION_LIMIT);
    let mut result: Vec<CacheData> = Vec::with_capacity(ids.len());
    for ids in id_chunks {
        @%- if def.dummy_always_joinable() %@
        let mut v = ids
            .iter()
            .map(|id| {
                let id = id.clone();
                CacheData {
                    @{- def.primaries()|fmt_join("
                    {ident}: id.{index}{raw_to_inner},", "") }@
                    ..Default::default()
                }
            })
            .collect();
        result.append(&mut v);
        @%- else %@
        let v = ___find_many(conn, CacheData::_sql_cols(@{ is_mysql_str }@), ids, TrashMode::With, None, Default::default()).await?;
        let mut v: sqlx::Result<Vec<_>> = v.iter().map(|(v, _)|CacheData::from_row(v)).collect();
        result.append(&mut v?);
        @%- endif %@
    }
    let time = MSec::now();
    let list = result.into_iter().map(|v| CacheWrapper::from_inner(v, conn.shard_id(), time)).collect();
    Ok(list)
}

async fn __find_many_from_cache<I, T>(conn: &mut DbConn, ids: I, joiner: Option<Box<Joiner_>>) -> Result<Vec<_@{ pascal_name }@Cache>>
where
    I: IntoIterator<Item = T>,
    T: Into<InnerPrimary>,
{
    let shard_id = conn.shard_id();
    let ids: Vec<_> = ids.into_iter().map(|id| PrimaryHasher(id.into(), shard_id)).collect();
    ___find_many_from_cache(conn, ids, joiner).await
}
#[allow(clippy::collapsible_if)]
async fn ___find_many_from_cache(conn: &mut DbConn, ids: Vec<PrimaryHasher>, joiner: Option<Box<Joiner_>>) -> Result<Vec<_@{ pascal_name }@Cache>> {
    let mut list: Vec<_@{ pascal_name }@Cache> = Vec::new();
    let mut rest_ids = Vec::new();
    let shard_id = conn.shard_id();
    let ctx_no = conn.ctx_no();
    let hash_and_id: Vec<_> = ids.into_iter().map(|id| (id.hash_val(shard_id), id)).collect();
    let cache_map = Cache::get_many::<CacheWrapper>(&hash_and_id.iter().map(|(hash, id)| *hash).collect(), shard_id, USE_FAST_CACHE).await;
    for (hash, id) in hash_and_id {
        if let Some(obj) = cache_map.get(&hash).filter(|o| InnerPrimary::from(*o) == id.0) {
            if obj._has_join(&joiner) {
                list.push((obj.clone(), Default::default()).into());
            } else {
                rest_ids.push(id);
            }
        } else {
            rest_ids.push(id);
        }
    }
    if !rest_ids.is_empty() {
        if rest_ids.len() <= 10 {
            for id in rest_ids.iter() {
                BULK_FETCH_QUEUE.get().unwrap()[shard_id as usize].push(id.0.clone());
            }
        }
        let mut conn = DbConn::_new_with_ctx(ctx_no, shard_id);
        conn.begin_cache_tx().await?;
        let mut rest_ids2 = FxHashSet::with_capacity_and_hasher(rest_ids.len() * 2, Default::default());
        for id in rest_ids.into_iter() {
            if let Some(obj) = Cache::get_from_memory::<CacheWrapper>(&id, shard_id, USE_FAST_CACHE).await.filter(|o| InnerPrimary::from(o) == id.0) {
                if obj._has_join(&joiner) {
                    list.push((obj, Default::default()).into());
                } else {
                    rest_ids2.insert(id);
                }
            } else {
                rest_ids2.insert(id);
            }
        }
        if !rest_ids2.is_empty() {
            let mut ids = FxHashSet::with_capacity_and_hasher(
                (BULK_FETCH_QUEUE.get().unwrap()[shard_id as usize].len() + rest_ids2.len()) * 2,
                Default::default()
            );
            for id in &rest_ids2 {
                ids.insert(id.0.clone());
            }
            let mut len = ids.len();
            if len < 100 {
                while let Some(x) = BULK_FETCH_QUEUE.get().unwrap()[shard_id as usize].pop() {
                    ids.insert(x);
                    len += 1;
                    if len >= 100 {
                        break;
                    }
                }
            }
            let mut ids: Vec<InnerPrimary> = ids.drain().collect();
            #[allow(unused_mut)]
            let mut result = ___find_many_for_cache(&mut conn, &ids).await?;
            if let Some(joiner) = &joiner {
@{- def.relations_in_cache(Joinable::Join)|fmt_rel_join("
                if joiner.{rel_name}.is_some() {
                    CacheWrapper::fetch_{raw_rel_name}_for_vec(&mut result, &mut conn).await?;
                }", "") }@
            }
            let _lock = db::models::CACHE_UPDATE_LOCK.read().await;
            for v in result.into_iter() {
                let arc = Arc::new(v);
                let id = PrimaryHasher(InnerPrimary::from(&arc), shard_id);
                let sync = CACHE_RESET_SYNC.get().unwrap()[shard_id as usize].read().await;
                if *sync <= conn.cache_sync() {
                    @%- if def.versioned %@
                    let vw = VersionWrapper {
                        id: id.0.clone(),
                        shard_id,
                        time: MSec::default(),
                        version: 0,
                    };
                    if let Some(ver) = Cache::get_version::<VersionWrapper>(&vw, shard_id).await.filter(|o| o.id == id.0) {
                        if arc._inner.@{ version_col }@.greater_equal(ver.version) {
                            Cache::insert_long(&id, arc.clone(), USE_FAST_CACHE).await;
                        }
                    } else {
                        let cs = CacheSyncWrapper {
                            id: id.0.clone(),
                            shard_id,
                            time: MSec::default(),
                            sync: 0,
                        };
                        if let Some(cs) = Cache::get_version::<CacheSyncWrapper>(&cs, shard_id).await.filter(|o| o.id == id.0) {
                            if cs.sync <= conn.cache_sync() {
                                Cache::insert_long(&id, arc.clone(), USE_FAST_CACHE).await;
                            }
                        } else {
                            Cache::insert_long(&id, arc.clone(), USE_FAST_CACHE).await;
                        }
                    }
                    @%- else %@
                    let cs = CacheSyncWrapper {
                        id: id.0.clone(),
                        shard_id,
                        time: MSec::default(),
                        sync: 0,
                    };
                    if let Some(cs) = Cache::get_version::<CacheSyncWrapper>(&cs, shard_id).await.filter(|o| o.id == id.0) {
                        if cs.sync <= conn.cache_sync() {
                            Cache::insert_long(&id, arc.clone(), USE_FAST_CACHE).await;
                        }
                    } else {
                        Cache::insert_long(&id, arc.clone(), USE_FAST_CACHE).await;
                    }
                    @%- endif %@
                }
                if rest_ids2.contains(&id) {
                    list.push((arc, Default::default()).into());
                }
            }
        }
        conn.release_cache_tx();
    }
    _@{ pascal_name }@Joiner::join(&mut list, conn, joiner).await?;
    Ok(list)
}
@%- endif %@

#[allow(clippy::needless_borrow)]
async fn __find_optional(conn: &mut DbConn, sql_cols: &str, id: InnerPrimary, trash_mode: TrashMode, filter: Option<Filter_>, filter_flag: BTreeMap<&'static str, Filter_>) -> Result<Option<(DbRow, BTreeMap<&'static str, bool>)>>
{
    let filter_flag_names: Vec<_> = filter_flag.keys().cloned().collect();
    let sql_cols = write_filter_flag(sql_cols, &filter_flag, conn.shard_id());
    let mut filter_str = Filter_::write_where(&filter, trash_mode, TRASHED_SQL, NOT_TRASHED_SQL, ONLY_TRASHED_SQL, conn.shard_id());
    if filter_str.is_empty() {
        filter_str = "WHERE".to_string();
    } else {
        filter_str.push_str(" AND ");
    }
    let sql = format!(r#"SELECT {} FROM @{ table_name|db_esc }@ as _t1 {filter_str} @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join("{col_esc}={placeholder}", " AND ") }@"#, sql_cols);
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let mut query = sqlx::query(&sql);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    for (_name, filter) in filter_flag {
        query = filter.bind_to_query(query);
    }
    if let Some(c) = filter {
        query = c.bind_to_query(query);
    }
    @{- def.primaries()|fmt_join("
    query = query.bind(id.{index}{bind_as});", "") }@
    Ok(crate::misc::fetch!(conn, query, fetch_optional)
        .map(|row| {
            use sqlx::Row;
            let mut flags = BTreeMap::new();
            for name in filter_flag_names.clone() {
                flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
            }
            (row, flags)
        }))
}
@%- if def.use_cache() %@

async fn __find_optional_from_cache<T>(conn: &mut DbConn, id: T, joiner: Option<Box<Joiner_>>) -> Result<Option<_@{ pascal_name }@Cache>>
where
    T: Into<InnerPrimary>,
{
    let id: InnerPrimary = id.into();
    let mut result = __find_many_from_cache(conn, [id], joiner).await?;
    Ok(result.pop())
}
@%- endif %@
@%- if !def.disable_update() %@

#[allow(clippy::needless_borrow)]
async fn __find_for_update(conn: &mut DbConn, id: &InnerPrimary, trash_mode: TrashMode, filter: Option<Filter_>, filter_flag: BTreeMap<&'static str, Filter_>) -> Result<Option<(Data, BTreeMap<&'static str, bool>)>> {
    let filter_flag_names: Vec<_> = filter_flag.keys().cloned().collect();
    let sql_cols = write_filter_flag(Data::_sql_cols(@{ is_mysql_str }@), &filter_flag, conn.shard_id());
    let mut filter_str = Filter_::write_where(&filter, trash_mode, TRASHED_SQL, NOT_TRASHED_SQL, ONLY_TRASHED_SQL, conn.shard_id());
    if filter_str.is_empty() {
        filter_str = "WHERE".to_string();
    } else {
        filter_str.push_str(" AND ");
    }
    let sql = format!(r#"SELECT {} FROM @{ table_name|db_esc }@ as _t1 {filter_str} @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join("{col_esc}={placeholder}", " AND ") }@ FOR UPDATE"#, sql_cols);
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let mut query = sqlx::query(&sql);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    for (_name, filter) in filter_flag {
        query = filter.bind_to_query(query);
    }
    if let Some(c) = filter {
        query = c.bind_to_query(query);
    }
    @{- def.primaries()|fmt_join("
    query = query.bind(id.{index}{bind_as});", "") }@
    let result = if conn.wo_tx() {
        query.fetch_optional(conn.acquire_writer().await?.as_mut()).await?
    } else {
        query.fetch_optional(conn.get_tx().await?.as_mut()).await?
    };
    Ok(result.map(|row| {
        use sqlx::Row;
        let mut flags = BTreeMap::new();
        for name in filter_flag_names.clone() {
            flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
        }
        Data::from_row(&row).map(|v| (v, flags))
    }).transpose()?)
}

#[allow(clippy::needless_borrow)]
async fn __find_many_for_update(conn: &mut DbConn, ids: &[InnerPrimary], trash_mode: TrashMode, joiner: Option<Box<Joiner_>>, filter: Option<Filter_>, filter_flag: BTreeMap<&'static str, Filter_>) -> Result<Vec<_@{ pascal_name }@Updater>>
{
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let filter_flag_names: Vec<_> = filter_flag.keys().cloned().collect();
    let sql_cols = write_filter_flag(Data::_sql_cols(@{ is_mysql_str }@), &filter_flag, conn.shard_id());
    let mut list: Vec<__Updater__> = Vec::with_capacity(ids.len());
    let mut filter_str = Filter_::write_where(&filter, trash_mode, TRASHED_SQL, NOT_TRASHED_SQL, ONLY_TRASHED_SQL, conn.shard_id());
    if filter_str.is_empty() {
        filter_str = "WHERE".to_string();
    } else {
        filter_str.push_str(" AND ");
    }
    let id_chunks = ids.chunks(IN_CONDITION_LIMIT);
    for ids in id_chunks {
        let q = "@{ def.primaries()|fmt_join_with_paren("{placeholder}", ",") }@,".repeat(ids.len());
        let sql = format!(
            r#"SELECT {} FROM @{ table_name|db_esc }@ {filter_str} @{ def.primaries()|fmt_join_with_paren("{col_esc}", ",") }@ in ({}) FOR UPDATE;"#,
            sql_cols,
            &q[0..q.len() - 1],
        );
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        for (_name, filter) in filter_flag.clone() {
            query = filter.bind_to_query(query);
        }
        if let Some(c) = filter.clone() {
            query = c.bind_to_query(query);
        }
        for id in ids {
            @{- def.primaries()|fmt_join("
            query = query.bind(id.{index}{bind_as});", "") }@
        }
        let result = if conn.wo_tx() {
            query.fetch_all(conn.acquire_writer().await?.as_mut()).await?
        } else {
            query.fetch_all(conn.get_tx().await?.as_mut()).await?
        };
        let result: sqlx::Result<Vec<_>> = result.into_iter().map(|row| {
            use sqlx::Row;
            let mut flags = BTreeMap::new();
            for name in filter_flag_names.clone() {
                flags.insert(name, row.get::<Option<bool>, &str>(name).unwrap_or_default());
            }            
            Data::from_row(&row).map(|v| (v, flags))}).collect();
        result?
            .into_iter()
            .map(__Updater__::from)
            .for_each(|obj| list.push(obj));
    }
    _@{ pascal_name }@Joiner::join(&mut list, conn, joiner).await?;
    Ok(list)
}
@%- endif %@

async fn __save(conn: &mut DbConn, obj: _@{ pascal_name }@Updater, rel_hash: u64) -> Result<Option<_@{ pascal_name }@>> {
    let (obj, cache_msg) = ___save(conn, obj, rel_hash).await?;
    @%- if !config.force_disable_cache %@
    @%- if def.act_as_job_queue() %@
    if let Some(cache_msg) = cache_msg {
        conn.push_cache_op(cache_msg.wrap()).await;
    }
    @%- else if def.clear_all_cache_on_update() %@
    conn.clear_all_cache = true;
    @%- else %@
    if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
        if let Some(cache_msg) = cache_msg {
            @%- if def.disable_insert_cache_propagation %@
            let internal = matches!(cache_msg, CacheOp::Insert { .. });
            conn.push_cache_op_to(cache_msg.wrap(), internal).await;
            @%- else %@
            conn.push_cache_op(cache_msg.wrap()).await;
            @%- endif %@
        }
    }
    @%- endif %@
    @%- endif %@
    Ok(obj)
}

#[allow(clippy::type_complexity)]
pub fn ___save(
    conn: &mut DbConn,
    obj: _@{ pascal_name }@Updater,
    rel_hash: u64,
) -> BoxFuture<'_, Result<(Option<_@{ pascal_name }@>, Option<CacheOp>)>>
{
    async move {
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
        if !conn.clear_all_cache {
            @{- def.cache_owners|fmt_cache_owners("
            if rel_hash != {rel_hash} {
                _invalidate_{raw_rel_name}(conn, &obj._data).await;
            }") }@
        }
        @%- endif %@
        @%- if !def.disable_delete() %@
        if obj.will_be_deleted() {
            if obj.is_new() || obj.has_been_deleted() {
                return Ok((None, None));
            }
            let cache_msg = __delete(conn, obj).await?;
            return Ok((None, cache_msg));
        }
        @%- endif %@
        @%- if !def.disable_update() %@
        if obj.is_new() && obj._upsert {
            let (obj, cache_msg) = __save_upsert(conn, obj).await?;
            Ok((Some(obj), cache_msg))
        } else if obj.is_new() {
            let (obj, cache_msg) = __save_insert(conn, obj, @%- if def.overwrite_on_relation_save() %@ rel_hash > 0 @%- else %@ rel_hash == 1 @%- endif %@).await?;
            Ok((Some(obj), Some(cache_msg)))
        } else {
            let (obj, cache_msg) = __save_update(conn, obj).await?;
            @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
            if !conn.clear_all_cache {
                @{- def.cache_owners|fmt_cache_owners("
                if rel_hash != {rel_hash} {
                    if let Some(v) = _base_filter_::repositories::{base_mod}::RelFk{rel_name_pascal}::get_fk(&obj._inner) {
                        if _base_filter_::repositories::{base_mod}::USE_CACHE || _base_filter_::repositories::{base_mod}::ENABLE_ALL_ROWS_CACHE || _base_filter_::repositories::{base_mod}::ENABLE_UPDATE_NOTICE {
                            let id: _base_filter_::repositories::{base_mod}::Primary = v.into();
                            let id: _base_filter_::repositories::{base_mod}::InnerPrimary = (&id).into();
                            conn.push_cache_op(_base_filter_::repositories::{base_mod}::CacheOp::Invalidate{id, shard_id: conn.shard_id()}.wrap()).await;
                        }
                    }
                }") }@
            }
            @%- endif %@
            Ok((Some(obj), cache_msg))
        }
        @%- else %@
        if obj.is_new() {
            let (obj, cache_msg) = __save_insert(conn, obj, @%- if def.overwrite_on_relation_save() %@ rel_hash > 0 @%- else %@ rel_hash == 1 @%- endif %@).await?;
            Ok((Some(obj), Some(cache_msg)))
        } else {
            anyhow::bail!("Update is disabled.");
        }
        @%- endif %@
    }.boxed()
}

#[allow(clippy::unnecessary_cast)]
async fn __save_insert(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater, overwrite: bool) -> Result<(_@{ pascal_name }@, CacheOp)> {
    let sql = if overwrite {
        @%- if config.is_mysql() %@
        r#"INSERT INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@) AS new ON DUPLICATE KEY UPDATE @{ def.non_primaries_except_created_at()|fmt_join("{col_esc}=new.{col_esc}", ",") }@@{ def.auto_inc()|fmt_join(",{col_esc}=LAST_INSERT_ID({col_esc})", "") }@;"#
        @%- else %@
        r#"INSERT INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@) ON CONFLICT @{ def.upsert_conflict_target() }@ DO UPDATE SET @{ def.non_primaries_except_created_at()|fmt_join("{col_esc}=excluded.{col_esc}", ",") }@@{ def.auto_inc()|fmt_join(" RETURNING xmax,{col_esc}", "") }@;"#
        @%- endif %@
    } else {
        r#"INSERT INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@)@% if !config.is_mysql() %@@{ def.auto_inc()|fmt_join(" RETURNING xmax,{col_esc}", "") }@@% endif %@;"#
    };
    @%- if !config.is_mysql() %@
    let sql = &senax_common::convert_mysql_placeholders_to_postgresql(sql);
    @%- endif %@
    let query = bind_to_query(sqlx::query(sql), &obj._data);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    let (_, _last_insert_id) = conn.execute@{ def.auto_inc()|fmt_join("_with_last_insert_id", "") }@(query).await.context(err::ErrorTable(TABLE_NAME))?;
@{- def.auto_inc()|fmt_join("
    if obj._data.{ident} == 0 {
        obj._data.{ident} = _last_insert_id as {inner};
    }", "") }@
    debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "insert", ctx = conn.ctx_no(); "{}", &obj);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
    let mut obj2: _@{ pascal_name }@ = (obj._data.clone(), BTreeMap::default()).into();
    let mut update_cache = true;

    @{- def.non_primaries()|fmt_join_cache_or_not("", "
    obj._data.{ident} = Default::default();", "") }@
    @%- if def.act_as_job_queue() %@
    let cache_msg = CacheOp::Queued;
    @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @%- else if !config.force_disable_cache && !def.clear_all_cache_on_update() %@
    let cache_msg = CacheOp::Insert {
        overwrite,
        shard_id: conn.shard_id(),
        data: obj._data,
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?,", "") }@
    };
@{- def.relations_one_uncached(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
@{- def.relations_many_uncached(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @%- else %@
    let cache_msg = CacheOp::None;
    @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @%- endif %@
    Ok((obj2, cache_msg))
}

@%- if !def.disable_update() || !def.disable_delete() %@

#[allow(unused_mut)]
#[allow(unused_assignments)]
#[allow(clippy::needless_borrow)]
#[allow(clippy::unnecessary_cast)]
async fn __save_update(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<(_@{ pascal_name }@, Option<CacheOp>)> {
    let id = InnerPrimary::from(&obj);
    let mut update_cache = false; // To distinguish from updates that do not require cache updates
    if obj.is_updated() {
        let mut vec: Vec<String> = Vec::new();
        @{- def.non_primaries_except_invisible_and_read_only(false)|fmt_join_cache_or_not("
        crate::misc::assign_sql!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, update_cache, \"{placeholder}\");", "
        crate::misc::assign_sql_no_cache_update!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, \"{placeholder}\");", "") }@
        @%- if config.is_mysql() %@
        @%- if def.versioned %@
        vec.push(r#"@{ version_col|db_esc }@ = LAST_INSERT_ID(IF(@{ version_col|db_esc }@ < 4294967295, @{ version_col|db_esc }@ + 1, 0))"#.to_string());
        @%- endif %@
        @%- if def.counter_field.is_some() %@
        vec.push(r#"@{ def.get_counter_field_col()|db_esc }@ = LAST_INSERT_ID(@{ def.get_counter_field_col()|db_esc }@)"#.to_string());
        @%- endif %@
        @%- else %@
        @%- if def.versioned %@
        vec.push(r#"@{ version_col|db_esc }@ = CASE WHEN @{ version_col|db_esc }@ < 2147483647 THEN @{ version_col|db_esc }@ + 1 ELSE 0 END"#.to_string());
        @%- endif %@
        @%- endif %@
        @%- if def.versioned %@
        let sql = format!(r#"UPDATE @{ table_name|db_esc }@ SET {} WHERE @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join("{col_esc}={placeholder}", " AND ") }@ AND @{ version_col|db_esc }@=?@% if !config.is_mysql() %@@% if def.versioned %@ RETURNING xmax,@{ version_col|db_esc }@@% endif %@@% if def.counter_field.is_some() %@ RETURNING xmax,@{ def.get_counter_field_col()|db_esc }@@% endif %@@% endif %@;"#, &vec.join(","));
        @%- else %@
        let sql = format!(r#"UPDATE @{ table_name|db_esc }@ SET {} WHERE @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join("{col_esc}={placeholder}", " AND ") }@@% if !config.is_mysql() %@@% if def.versioned %@ RETURNING xmax,@{ version_col|db_esc }@@% endif %@@% if def.counter_field.is_some() %@ RETURNING xmax,@{ def.get_counter_field_col()|db_esc }@@% endif %@@% endif %@;"#, &vec.join(","));
        @%- endif %@
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        @{- def.non_primaries_except_invisible_and_read_only(false)|fmt_join("
        for _n in 0..obj._op.{ident}.get_bind_num({may_null}) {
            query = query.bind(obj._update.{ident}{bind_as});
        }","") }@
        query = query@{ def.primaries()|fmt_join(".bind(id.{index}{bind_as})", "") }@;
        @%- if def.versioned %@
        query = query.bind(obj._data.@{ version_col }@);
        @%- endif %@
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "update", ctx = conn.ctx_no(); "{}", &obj);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
        let (rows_affected, _last_insert_id) = conn.execute@% if def.versioned %@_with_last_insert_id@% endif %@(query).await.context(err::ErrorTable(TABLE_NAME))?;
        if rows_affected == 0 {
            anyhow::bail!(err::RowNotFound::new(TABLE_NAME, id.to_string()));
        }
        @%- if def.versioned %@
        obj._data.@{ version_col }@ = _last_insert_id as @{ config.u32() }@;
        obj._update.@{ version_col }@ = _last_insert_id as @{ config.u32() }@;
        obj._op.@{ version_col }@ = Op::Set;
        @%- endif %@
        @%- if def.counter_field.is_some() %@
        if obj._op.@{ def.get_counter_field() }@ == Op::Add {
            obj._op.@{ def.get_counter_field() }@ = Op::Max;
            obj._update.@{ def.get_counter_field() }@ = _last_insert_id.try_into().unwrap_or(@{ def.get_counter_field_type() }@::MAX);
        }
        @%- endif %@
    }
    let mut obj2: _@{ pascal_name }@ = (obj._data, BTreeMap::default()).into();
    @{- def.non_primaries()|fmt_join_cache_or_not("", "
    obj._op.{ident} = Op::None;
    obj._update.{ident} = Default::default();", "") }@
    @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && !def.act_as_job_queue() %@
    let mut cache_msg = Some(CacheOp::Update {
        id,
        shard_id: conn.shard_id(),
        update:obj._update,
        op: obj._op,
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n            _{rel_name}: save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?,", "") }@
    });
    if !update_cache {
        cache_msg = None;
    }
@{- def.relations_one_uncached(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
@{- def.relations_many_uncached(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @%- else %@
    let cache_msg = None;
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("\n        save_{rel_name}(conn, &mut obj2, obj.{rel_name}, &mut update_cache).await?;", "") }@
    @%- endif %@
    Ok((obj2, cache_msg))
}
@%- endif %@
@%- if !def.disable_update() %@

fn assign_non_primaries(obj: &_@{ pascal_name }@Updater) -> (Vec<String>, bool) {
    let mut update_cache = false;
    let mut vec: Vec<String> = Vec::new();
    @{- def.non_primaries_except_read_only()|fmt_join_cache_or_not("
    crate::misc::assign_sql!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, update_cache, \"{placeholder}\");", "
    crate::misc::assign_sql_no_cache_update!(obj, vec, {ident}, r#\"{col_esc}\"#, {may_null}, \"{placeholder}\");", "") }@
    (vec, update_cache)
}

fn bind_non_primaries<'a>(obj: &'a _@{ pascal_name }@Updater, mut query: Query<'a, DbType, DbArguments>, _sql: &'a str) -> Query<'a, DbType, DbArguments> {
    @{- def.non_primaries_except_read_only()|fmt_join("
    for _n in 0..obj._op.{ident}.get_bind_num({may_null}) {
        query = query.bind(obj._update.{ident}{bind_as});
    }","") }@
    query
}

#[allow(unused_assignments)]
#[allow(clippy::unnecessary_cast)]
async fn __save_upsert(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<(_@{ pascal_name }@, Option<CacheOp>)> {
    let (mut vec, _) = assign_non_primaries(&obj);
    @%- if config.is_mysql() %@
    @%- if def.versioned %@
    vec.push(r#"@{ version_col|db_esc }@ = LAST_INSERT_ID(IF(@{ version_col|db_esc }@ < 4294967295, @{ version_col|db_esc }@ + 1, 0))"#.to_string());
    @%- endif %@
    @%- if def.counter_field.is_some() %@
    vec.push(r#"@{ def.get_counter_field_col()|db_esc }@ = LAST_INSERT_ID(@{ def.get_counter_field_col()|db_esc }@)"#.to_string());
    @%- endif %@
    @%- else %@
    @%- if def.versioned %@
    vec.push(r#"@{ version_col|db_esc }@ = CASE WHEN @{ table_name|db_esc }@.@{ version_col|db_esc }@ < 2147483647 THEN @{ table_name|db_esc }@.@{ version_col|db_esc }@ + 1 ELSE 0 END"#.to_string());
    @%- endif %@
    @%- endif %@
    let sql = format!(r#"INSERT INTO @{ table_name|db_esc }@ 
        (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) 
        VALUES (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@) @{ config.db_type_switch("ON DUPLICATE KEY UPDATE", "ON CONFLICT ") }@@{ def.upsert_conflict_target() }@@{ config.db_type_switch("", " DO UPDATE SET") }@ {}@% if !config.is_mysql() %@@% if def.versioned %@ RETURNING xmax,@{ version_col|db_esc }@@% endif %@@% if def.counter_field.is_some() %@ RETURNING xmax,@{ def.get_counter_field_col()|db_esc }@@% endif %@@% endif %@;"#, &vec.join(","));
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let query = bind_to_query(sqlx::query(&sql), &obj._data);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    let query = bind_non_primaries(&obj, query, &sql);
    let (rows_affected, _last_insert_id) = conn.execute@% if def.versioned || def.counter_field.is_some() %@_with_last_insert_id@% endif %@(query).await.context(err::ErrorTable(TABLE_NAME))?;
    debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "upsert", ctx = conn.ctx_no(); "{}", &obj);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
    if rows_affected == 1 {
        @{- def.auto_inc()|fmt_join("
        if obj._data.{ident} == 0 {
            obj._data.{ident} = _last_insert_id as {inner};
        }", "") }@
        let mut obj2: _@{ pascal_name }@ = (obj._data.clone(), BTreeMap::default()).into();
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && !def.act_as_job_queue() %@
        let cache_msg = Some(CacheOp::Insert {
            overwrite: false,
            shard_id: conn.shard_id(),
            data: obj._data,
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
        });
        @%- else %@
        let cache_msg = None;
        @%- endif %@
        Ok((obj2, cache_msg))
    } else if rows_affected == 2 {
        @%- if def.versioned %@
        obj._data.@{ version_col }@ = _last_insert_id as @{ config.u32() }@;
        obj._update.@{ version_col }@ = _last_insert_id as @{ config.u32() }@;
        obj._op.@{ version_col }@ = Op::Set;
        @%- endif %@
        @%- if def.counter_field.is_some() %@
        if obj._op.@{ def.get_counter_field() }@ == Op::Add {
            obj._op.@{ def.get_counter_field() }@ = Op::Max;
            obj._update.@{ def.get_counter_field() }@ = _last_insert_id.try_into().unwrap_or(@{ def.get_counter_field_type() }@::MAX);
        }
        @%- endif %@
        let id = InnerPrimary::from(&obj);
        let mut obj2: _@{ pascal_name }@ = (obj._data, BTreeMap::default()).into();
        @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && !def.act_as_job_queue() %@
        let mut cache_msg = Some(CacheOp::Update {
            id,
            shard_id: conn.shard_id(),
            update:obj._update,
            op: obj._op,
@{- def.relations_one_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
@{- def.relations_many_cache(Joinable::Join, false)|fmt_rel_join("\n                _{rel_name}: None,", "") }@
        });
        @%- else %@
        let cache_msg = None;
        @%- endif %@
        Ok((obj2, cache_msg))
    } else {
        let mut obj2: _@{ pascal_name }@ = (obj._data, BTreeMap::default()).into();
        Ok((obj2, None))
    }
}
@%- endif %@
@%- if !def.disable_update() %@

#[allow(unused_mut)]
async fn __update_many(conn: &mut DbConn, ids: Vec<InnerPrimary>, mut obj: __Updater__) -> Result<u64> {
    if !obj.is_updated() {
        return Ok(0);
    }
    if ids.is_empty() {
        return Ok(0);
    }
    let mut rows_affected = 0;
    for ids in ids.chunks(IN_CONDITION_LIMIT) {
        rows_affected += ___update_many(conn, ids, &obj).await?;
    }
    debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "update_many", ctx = conn.ctx_no(), ids = primaries_to_str(&ids); "{}", &obj);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &obj);
    @%- if !config.force_disable_cache %@
    @%- if def.act_as_job_queue() %@
    @%- else if def.clear_all_cache_on_update() %@
    conn.clear_all_cache = true;
    @%- else %@
    if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
        @{- def.non_primaries()|fmt_join_cache_or_not("", "
        obj._op.{ident} = Op::None;", "") }@
        let mut data_list = Vec::new();
        @%- if def.non_primaries_addable().len() > 0 %@
        let has_add = false
            @{- def.non_primaries_addable()|fmt_join_cache_or_not("
            || obj._op.{ident} == Op::Add || obj._op.{ident} == Op::Sub", "", "") }@;
        if has_add {
            for ids in ids.chunks(IN_CONDITION_LIMIT) {
                _repo_::find_many_for_update(conn, ids.iter(), None, None, None).await?.into_iter()
                    .for_each(|v| {
                        let mut data = v._data;
                        @{- def.non_primaries_addable()|fmt_join_cache_or_not("
                        if obj._op.{ident} != Op::Add && obj._op.{ident} != Op::Sub {
                            data.{ident} = obj._update.{ident}.clone();
                        }", "", "") }@;
                        @{- def.non_primaries()|fmt_join_cache_or_not("", "
                        data.{ident} = Default::default();", "") }@
                        data_list.push(data);
                    });
            }
        }
        @%- endif %@
        @{- def.non_primaries_addable()|fmt_join_cache_or_not("
        if obj._op.{ident} == Op::Add {
            obj._op.{ident} = Op::Max;
        } else if obj._op.{ident} == Op::Sub {
            obj._op.{ident} = Op::Min;
        }", "", "") }@
        @{- def.non_primaries()|fmt_join_cache_or_not("", "
        obj._update.{ident} = Default::default();", "") }@
        let cache_msg = CacheOp::UpdateMany {
            ids,
            shard_id: conn.shard_id(),
            update: obj._update,
            data_list,
            op: obj._op,
        };
        conn.push_cache_op(cache_msg.wrap()).await;
    }
    @%- endif %@
    @%- endif %@
    Ok(rows_affected)
}
@%- endif %@
@%- if !def.disable_update() %@

#[allow(unused_assignments)]
#[allow(unused_mut)]
#[allow(clippy::needless_borrow)]
async fn ___update_many(conn: &mut DbConn, ids: &[InnerPrimary], obj: &__Updater__) -> Result<u64> {
    if ids.is_empty() {
        return Ok(0);
    }
    let (mut vec, _) = assign_non_primaries(obj);
    let q = "@{ def.primaries()|fmt_join_with_paren("{placeholder}", ",") }@,".repeat(ids.len());
    let sql = format!(r#"UPDATE @{ table_name|db_esc }@ SET {} WHERE @{ def.inheritance_cond(" AND ") }@@{ def.primaries()|fmt_join_with_paren("{col_esc}", ",") }@ in ({})"#, &vec.join(","), &q[0..q.len() - 1]);
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let query = sqlx::query(&sql);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    let mut query = bind_non_primaries(&obj, query, &sql);
    for id in ids {
        @{- def.primaries()|fmt_join("
        query = query.bind(id.{index}{bind_as});", "") }@
    }
    let result = if conn.wo_tx() {
        query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
    } else {
        query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
    };
    Ok(result.rows_affected())
}
@%- endif %@

async fn _bulk_insert(conn: &mut DbConn, mut list: Vec<_@{ pascal_name }@Updater>, ignore: bool, replace: bool) -> Result<()> {
    let mut vec: Vec<ForInsert> = Vec::with_capacity(list.len());
    for mut obj in list.into_iter() {
        ensure!(obj.is_new(), "The obj is not new.");
        obj.__validate()?;
        obj.__set_default_value(conn).await?;
        vec.push(obj.into());
    }
    __bulk_insert(conn, &vec, ignore, replace, false).await
}

async fn __bulk_insert(conn: &mut DbConn, list: &[ForInsert], ignore: bool, replace: bool, overwrite: bool) -> Result<()> {
    let result = ___bulk_insert(conn, list, ignore, replace, overwrite).await?;
    @%- if !config.force_disable_cache %@
    @%- if def.act_as_job_queue() %@
    conn.push_cache_op(CacheOp::Queued.wrap()).await;
    @%- else if def.clear_all_cache_on_update() %@
    conn.clear_all_cache = true;
    @%- else %@
    @%- if !config.force_disable_cache && !def.clear_all_cache_on_update() && def.cache_owners.len() > 0 %@
    if !conn.clear_all_cache {
        for list in &result {
            for obj in list.iter() {
                @{- def.cache_owners|fmt_cache_owners("
                _invalidate_{raw_rel_name}(conn, &obj._data).await;") }@
            }
        }
    }
    @%- endif %@
    if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
        for list in result {
            let cache_msg = CacheOp::BulkInsert {
                replace,
                overwrite,
                ignore, 
                shard_id: conn.shard_id(),
                list,
            };
            conn.push_cache_op_to(cache_msg.wrap(), @{ def.disable_insert_cache_propagation }@).await;
        }
    }
    @%- endif %@
    @%- endif %@
    Ok(())
}

pub async fn ___bulk_insert(conn: &mut DbConn, list: &[ForInsert], ignore: bool, replace: bool, overwrite: bool) -> Result<Vec<Vec<ForInsert>>> {
    if list.is_empty() {
        return Ok(Vec::new());
    }
    let chunks = crate::misc::split_by_weight(list, *BULK_INSERT_MAX_SIZE.get().unwrap(), @{ 15000 / def.all_fields_except_read_only_and_auto_inc().len() }@, |v| v._data._size());
    let mut result = Vec::new();
    for chunk in chunks {
        result.push(____bulk_insert(conn, chunk, ignore, replace, overwrite).await?);
    }
    Ok(result)
}

#[allow(clippy::needless_borrow)]
#[allow(clippy::unnecessary_cast)]
#[allow(clippy::needless_if)]
#[allow(clippy::unwrap_or_default)]
fn ____bulk_insert<'a>(conn: &'a mut DbConn, list: &'a [ForInsert], ignore: bool, replace: bool, overwrite: bool) -> BoxFuture<'a, Result<Vec<ForInsert>>> {
    async move {
        if list.is_empty() {
            return Ok(Vec::new());
        }
        @%- if config.is_mysql() %@
        const SQL_NORMAL: &str = r#"INSERT "#; 
        const SQL_IGNORE: &str = r#"INSERT IGNORE "#; 
        const SQL_REPLACE: &str = r#"REPLACE "#; 
        const SQL1: &str = r#"INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES "#;
        const SQL2: &str = r#"(@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@)"#;
        const SQL3: &str = r#" AS new ON DUPLICATE KEY UPDATE @{ def.non_primaries_except_created_at()|fmt_join("{col_esc}=new.{col_esc}", ",") }@"#;
        let mut sql = String::with_capacity(SQL_IGNORE.len() + SQL1.len() + (SQL2.len() + 1) * list.len() + SQL3.len() + 200);
        if ignore {
            sql.push_str(SQL_IGNORE);
        } else if replace {
            sql.push_str(SQL_REPLACE);
        } else {
            sql.push_str(SQL_NORMAL);
        }
        sql.push_str(SQL1);
        sql.push_str(SQL2);
        for _i in 0..list.len() - 1 {
            sql.push(',');
            sql.push_str(SQL2);
        }
        if overwrite {
            sql.push_str(SQL3);
            @%- if def.versioned %@
            sql.push_str(r#",@{ version_col|db_esc }@ = IF(@{ version_col|db_esc }@ < 4294967295, @{ version_col|db_esc }@ + 1, 0)"#);
            @%- endif %@
        }
        @%- else %@
        const SQL1: &str = r#"INSERT INTO @{ table_name|db_esc }@ AS old (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES "#;
        const SQL2: &str = r#"(@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@)"#;
        const SQL_IGNORE: &str = r#" ON CONFLICT DO NOTHING"#;
        const SQL_REPLACE: &str = r#" ON CONFLICT @{ def.upsert_conflict_target() }@ DO UPDATE SET @{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}=excluded.{col_esc}", ",") }@"#;
        const SQL_OVERWRITE: &str = r#" ON CONFLICT @{ def.upsert_conflict_target() }@ DO UPDATE SET @{ def.non_primaries_except_created_at()|fmt_join("{col_esc}=excluded.{col_esc}", ",") }@"#;
        let mut sql = String::with_capacity(SQL1.len() + (SQL2.len() + 1) * list.len() + SQL_REPLACE.len() + 200);
        sql.push_str(SQL1);
        sql.push_str(SQL2);
        for _i in 0..list.len() - 1 {
            sql.push(',');
            sql.push_str(SQL2);
        }
        if ignore {
            sql.push_str(SQL_IGNORE);
        } else if replace {
            sql.push_str(SQL_REPLACE);
        } else if overwrite {
            sql.push_str(SQL_OVERWRITE);
            @%- if def.versioned %@
            sql.push_str(r#",@{ version_col|db_esc }@ = CASE WHEN old.@{ version_col|db_esc }@ < 2147483647 THEN old.@{ version_col|db_esc }@ + 1 ELSE 0 END"#);
            @%- endif %@
        }
            @%- if def.is_auto_inc() %@
        sql.push_str(r#" RETURNING @{ def.auto_inc()|fmt_join("{col_esc}", "") }@"#);
            @%- endif %@
        @%- endif %@
        @%- if !config.is_mysql() %@
        let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
        @%- endif %@
        let mut query = sqlx::query(&sql);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
        for data in list {
            query = bind_to_query(query, &data._data);
        }
    @%- if config.is_mysql() %@
        let result = if conn.wo_tx() {
            query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        } else {
            query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        };
        @{- def.auto_inc()|fmt_join("
        let mut id = result.last_insert_id() as {inner};", "") }@
    @%- else %@
        let result = if conn.wo_tx() {
            query.@% if def.is_auto_inc() %@fetch_all@% else %@execute@% endif %@(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        } else {
            query.@% if def.is_auto_inc() %@fetch_all@% else %@execute@% endif %@(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?
        };
        @{- def.auto_inc()|fmt_join("
        use sqlx::Row;
        let ids: Vec<_> = result.iter().map(|v| v.get::<{inner}, usize>(0)).collect();
        let mut ids_iter = ids.iter();", "") }@
    @%- endif %@
        let mut data_list = Vec::new();
        @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
        let mut _{rel_name} = Vec::new();
        let mut __{rel_name} = Vec::new();", "") }@
        @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
        let mut _{rel_name} = Vec::new();
        let mut __{rel_name} = Vec::new();", "") }@
        for row in list {
            debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "bulk_insert", ctx = conn.ctx_no(); "{}", &row);
            debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &row);
            let mut obj = row.clone();
    @%- if config.is_mysql() %@
            @{- def.auto_inc()|fmt_join("
            if obj._data.{ident} == 0 {
                obj._data.{ident} = id;
                // innodb_autoinc_lock_mode must be 0 or 1
                id += 1;
            }", "") }@
    @%- else %@
            @{- def.auto_inc()|fmt_join("
            obj._data.{ident} = *ids_iter.next().unwrap();", "") }@
    @%- endif %@
            @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
            obj.{rel_name}.map(|v| v.map(|v| {
                let mut v = v.as_ref().clone();
                RelFk{rel_name_pascal}::set_fk(&mut v, (&obj._data).into());
                _{rel_name}.push(v);
                __{rel_name}.push(row);
            }));", "") }@
            @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
            if let Some(v) = obj.{rel_name} {
                v.into_iter().for_each(|mut v| {
                    RelFk{rel_name_pascal}::set_fk(&mut v, (&obj._data).into());
                    _{rel_name}.push(v);
                });
                __{rel_name}.push(row);
            }", "") }@
            data_list.push((obj._data, obj._is_new));
        }
        if ignore || replace || overwrite {
            @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
            if !__{rel_name}.is_empty() {
                let filter = RelFil{rel_name_pascal}::in_filter(&__{rel_name});
                repo_{class_mod}::_repo_::query().filter(filter).delete(conn).await?;
            }", "") }@
            @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
            if !__{rel_name}.is_empty() {
                let filter = RelFil{rel_name_pascal}::in_filter(&__{rel_name});
                repo_{class_mod}::_repo_::query().filter(filter).delete(conn).await?;
            }", "") }@
        }
        @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
        let mut _{rel_name} = repo_{class_mod}::___bulk_insert(conn, &_{rel_name}, ignore, replace, overwrite).await?.into_iter().fold(FxHashMap::default(), |mut map, v| {
            for v in v {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&v._data) {
                    map.insert(id, v);
                }
            }
            map
        });", "") }@
        @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
        let mut _{rel_name} = repo_{class_mod}::___bulk_insert(conn, &_{rel_name}, ignore, replace, overwrite).await?.into_iter().fold(FxHashMap::default(), |mut map: FxHashMap<_, Vec<_>>, v| {
            for v in v {
                if let Some(id) = RelFk{rel_name_pascal}::get_fk(&v._data) {
                    map.entry(id).or_default().push(v);
                }
            }
            map
        });", "") }@
        let data_list = data_list.into_iter().map(|(v, n)| ForInsert {
            @{- def.relations_one(Joinable::Join, false)|fmt_rel_join("
            {rel_name}: Some(_{rel_name}.remove(&(&v).into()).map(Box::new)),", "") }@
            @{- def.relations_many(Joinable::Join, false)|fmt_rel_join("
            {rel_name}: _{rel_name}.remove(&(&v).into()),", "") }@
            _data: v,
            _is_new: n,
        }).collect();
        Ok(data_list)
    }.boxed()
}
@%- if !def.disable_update() %@

async fn __bulk_upsert(conn: &mut DbConn, list: &[Data], obj: &__Updater__) -> Result<()> {
    let chunks = crate::misc::split_by_weight(list, *BULK_INSERT_MAX_SIZE.get().unwrap(), @{ 15000 / def.all_fields_except_read_only_and_auto_inc().len() }@, |v| v._size());
    for chunk in chunks {
        ___bulk_upsert(conn, chunk, obj).await?;
    }
    Ok(())
}

#[allow(unused_assignments)]
#[allow(clippy::needless_borrow)]
async fn ___bulk_upsert(conn: &mut DbConn, list: &[Data], obj: &__Updater__) -> Result<()> {
    if list.is_empty() {
        return Ok(());
    }
    const SQL1: &str = r#"INSERT INTO @{ table_name|db_esc }@ (@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{col_esc}", ",") }@) VALUES "#;
    const SQL2: &str = r#"(@{ def.all_fields_except_read_only_and_auto_inc()|fmt_join("{placeholder}", ",") }@)"#;
    let mut sql = String::with_capacity(SQL1.len() + (SQL2.len() + 1) * list.len() + 100);
    sql.push_str(SQL1);
    sql.push_str(SQL2);
    for _i in 0..list.len() - 1 {
        sql.push(',');
        sql.push_str(SQL2);
    }
    let (mut vec, _) = assign_non_primaries(obj);
    @%- if def.versioned %@
    @%- if config.is_mysql() %@
    vec.push(r#"@{ version_col|db_esc }@ = IF(@{ version_col|db_esc }@ < 4294967295, @{ version_col|db_esc }@ + 1, 0)"#.to_string());
    @%- else %@
    vec.push(r#"@{ version_col|db_esc }@ = CASE WHEN @{ table_name|db_esc }@.@{ version_col|db_esc }@ < 2147483647 THEN @{ table_name|db_esc }@.@{ version_col|db_esc }@ + 1 ELSE 0 END"#.to_string());
    @%- endif %@
    @%- endif %@
    write!(sql, r#" @{ config.db_type_switch("ON DUPLICATE KEY UPDATE", "ON CONFLICT ") }@@{ def.upsert_conflict_target() }@@{ config.db_type_switch("", " DO UPDATE SET") }@ {}"#, &vec.join(","))?;
    @%- if !config.is_mysql() %@
    let sql = senax_common::convert_mysql_placeholders_to_postgresql(&sql);
    @%- endif %@
    let mut query = sqlx::query(&sql);
    debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", ctx = conn.ctx_no(), sql = &query.sql(); "query");
    for data in list {
        query = bind_to_query(query, data);
        debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "bulk_upsert", ctx = conn.ctx_no(); "{}", &data);
        debug!(target: "db_@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", "{:?}", &data);
    }
    debug!(target: "_db_update::@{ db|snake }@::@{ group_name|snake }@::@{ mod_name }@", op = "bulk_upsert_updater", ctx = conn.ctx_no(); "{}", obj);
    let query = bind_non_primaries(&obj, query, &sql);
    if conn.wo_tx() {
        query.execute(conn.acquire_writer().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
    } else {
        query.execute(conn.get_tx().await?.as_mut()).await.context(err::ErrorTable(TABLE_NAME))?;
    }
    @%- if !config.force_disable_cache %@
    @%- if def.act_as_job_queue() %@
    @%- else if def.clear_all_cache_on_update() %@
    conn.clear_all_cache = true;
    @%- else %@
    if !conn.clear_all_cache && (USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE) {
        let cache_msg = CacheOp::BulkUpsert {
            shard_id: conn.shard_id(),
            data_list: list.to_vec(),
            update: obj._update.clone(),
            op: obj._op.clone(),
        };
        conn.push_cache_op(cache_msg.wrap()).await;
    }
    @%- endif %@
    @%- endif %@
    Ok(())
}
@%- endif %@
@%- if !def.disable_delete() %@

#[allow(unused_mut)]
async fn __delete(conn: &mut DbConn, mut obj: _@{ pascal_name }@Updater) -> Result<Option<CacheOp>> {
    if obj.has_been_deleted() {
        return Ok(None);
    }
@{- def.soft_delete_tpl2("
    _repo_::force_delete(conn, obj).await?;
    Ok(None)","
    obj.mut_deleted_at().set(Some({val}.into()));
    let (_obj, cache_msg) = __save_update(conn, obj).await?;
    Ok(cache_msg)","
    obj.mut_deleted().set(true);
    let (_obj, cache_msg) = __save_update(conn, obj).await?;
    Ok(cache_msg)","
    let deleted = cmp::max(1, SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() as {u32});
    obj.mut_deleted().set(deleted);
    let (_obj, cache_msg) = __save_update(conn, obj).await?;
    Ok(cache_msg)")}@
}
@%- endif %@

#[allow(unused_mut)]
@%- if force_indexes.len() > 0 %@
#[inline(never)]
@%- endif %@
fn make_force_indexes(filter_digest: &str) -> Vec<&'static str> {
    let mut force_indexes: Vec<&'static str> = Vec::new();
    @%- for (cond, idx) in force_indexes %@
    if @{ cond }@ {
        force_indexes.push(@{ idx }@);
    }
    @%- endfor %@
    @%- if !force_indexes.is_empty() %@
    if !force_indexes.is_empty()@{ def.primaries()|fmt_join(" && filter_digest.contains(\"`{col}`\")", "") }@ {
        force_indexes.push("\"PRIMARY\"");
    }
    @%- endif %@
    force_indexes
}

#[allow(clippy::needless_borrow)]
fn bind_to_query<'a>(mut query: Query<'a, DbType, DbArguments>, data: &'a Data) -> Query<'a, DbType, DbArguments> {
    @{- def.all_fields_except_read_only_and_auto_inc()|fmt_join("
    query = query.bind(data.{ident}{bind_as});", "") }@
    query
}

pub(crate) async fn _seed(seed: &serde_yaml::Value, conns: &mut [DbConn]) -> Result<()> {
    if let Some(mapping) = seed.as_mapping() {
        for (name, factory) in mapping {
            let seed: _@{ pascal_name }@Factory = serde_yaml::from_str(&serde_yaml::to_string(&factory)?)?;
            let conn = &mut conns[0];
            let obj = seed.create();
            if let Some(obj) = _repo_::save(conn, obj).await? {
                @{- def.auto_inc_or_seq()|fmt_join("
                let id = obj._{raw_name}();
                if GENERATED_IDS.get().is_none() {
                    let _ = GENERATED_IDS.set(std::sync::RwLock::new(HashMap::new()));
                }
                let name = name.as_str().unwrap().to_string();
                GENERATED_IDS.get().unwrap().write().unwrap().insert(name, id);", "") }@
            }
        }
    }
    Ok(())
}
@{- def.relations_one(Joinable::Join, false)|fmt_rel_join("

async fn save_{rel_name}(
    conn: &mut DbConn,
    obj: &mut _{pascal_name},
    data: Option<Vec<rel_{class_mod}::{class}Updater>>,
    update_cache: &mut bool,
) -> Result<Option<Vec<rel_{class_mod}::CacheOp>>> {
    if let Some(list) = data {
        let mut msgs = Vec::new();
        let src = _save_{rel_name}(conn, (&*obj).into(), list);
        futures::pin_mut!(src);
        let mut vec = Vec::new();
        while let Some(value) = src.next().await.transpose()? {
            if let Some(obj2) = value.0 {
                vec.push(obj2);
            }
            if let Some(msg) = value.1 {
                msgs.push(msg);
                *update_cache = true;
            }
        }
        obj.{rel_name} = Some(vec.pop().map(Box::new));
        return Ok(Some(msgs));
    }
    Ok(None)
}", "") }@
@{- def.relations_many(Joinable::Join, false)|fmt_rel_join("

async fn save_{rel_name}(
    conn: &mut DbConn,
    obj: &mut _{pascal_name},
    data: Option<Vec<rel_{class_mod}::{class}Updater>>,
    update_cache: &mut bool,
) -> Result<Option<Vec<rel_{class_mod}::CacheOp>>> {
    if let Some(list) = data {
        let mut msgs = Vec::new();
        let src = _save_{rel_name}(conn, (&*obj).into(), list);
        futures::pin_mut!(src);
        let mut vec = Vec::new();
        while let Some(value) = src.next().await.transpose()? {
            if let Some(obj2) = value.0 {
                vec.push(obj2);
            }
            if let Some(msg) = value.1 {
                msgs.push(msg);
                *update_cache = true;
            }
        }
        obj.{rel_name} = Some(vec);
        return Ok(Some(msgs));
    }
    Ok(None)
}", "") }@
@{- def.relations_one_and_many(Joinable::Join, false)|fmt_rel_join("

fn _save_{rel_name}(
    conn: &mut DbConn,
    id: InnerPrimary,
    list: Vec<rel_{class_mod}::{class}Updater>,
) -> impl futures::Stream<
    Item = Result<(
        Option<rel_{class_mod}::{class}>,
        Option<rel_{class_mod}::CacheOp>,
    )>,
> + '_ {
    async_stream::try_stream! {
        let mut update_list = Vec::with_capacity(list.len());
        let mut insert_list = Vec::with_capacity(list.len());
        for row in list.into_iter() {
            if row.is_new() {
                insert_list.push(row);
            }else if row.will_be_deleted() {
                yield repo_{class_mod}::___save(conn, row, {rel_hash}).await?;
            } else{
                update_list.push(row);
            }
        }
        for mut row in update_list.into_iter() {
            RelFk{rel_name_pascal}::set_fk(&mut row._data, id.clone());
            RelFk{rel_name_pascal}::set_fk(&mut row._update, id.clone());
            RelCol{rel_name_pascal}::set_op_none(&mut row._op);
            yield repo_{class_mod}::___save(conn, row, {rel_hash}).await?;
        }
        for mut row in insert_list.into_iter() {
            RelFk{rel_name_pascal}::set_fk(&mut row._data, id.clone());
            yield repo_{class_mod}::___save(conn, row, {rel_hash}).await?;
        }
    }
}", "") }@
@{- def.cache_owners|fmt_cache_owners("
#[inline(never)]
async fn _invalidate_{raw_rel_name}(conn: &mut DbConn, data: &Data) {
    use _base_filter_::repositories::{base_mod}::{RelFk{rel_name_pascal}, USE_CACHE, ENABLE_ALL_ROWS_CACHE, ENABLE_UPDATE_NOTICE, Primary, CacheOp};
    if let Some(v) = RelFk{rel_name_pascal}::get_fk(data) {
        if USE_CACHE || ENABLE_ALL_ROWS_CACHE || ENABLE_UPDATE_NOTICE {
            let id: Primary = v.into();
            conn.push_cache_op(CacheOp::Invalidate{id: (&id).into(), shard_id: conn.shard_id()}.wrap()).await;
        }
    }
}") }@
@{-"\n"}@