pi_db 0.19.1

//! 数据库启动、表注册、管理入口与根事务树装配。
//!
//! 本模块位于公开数据库 API 与 `pi_async_transaction`/`pi_store` 之间：
//!
//! - [`crate::db::KVDBManagerBuilder`] 创建目录、加载内部 Meta 表和用户表，然后始终通过内部
//!   `try_repair` 重放未确认的根前导日志；
//! - [`crate::db::KVDBManager`] 共享 runtime、两阶段事务管理器、根 WAL logger、表注册表和事件通道；
//! - [`crate::db::KVDBTransaction`] 是公开事务句柄，应用层只能把
//!   [`crate::db::KVDBTransaction::RootTr`] 当作
//!   动作和生命周期入口，其余 variant 是事务树内部子节点；
//! - [`crate::db::RootTransaction`] 按首次触表顺序持有子事务，并把 prepare、根 WAL、子表发布和最终
//!   确认串成两阶段提交闭环。
//!
//! 当前 `close`、只读事务写入、timeout 和 prepare token 仍按实现事实记录，并不是最终或
//! 最佳设计。稳定事务/确认协议见 `docs/SEMANTIC_CONTRACTS.md#contract-transaction`，管理器
//! 当前契约由 `tests/manager_contract.rs` 验证，根生命周期由
//! `tests/root_transaction_lifecycle.rs` 验证。

use std::mem::swap;
use std::time::{Duration, Instant};
use std::convert::TryInto;
use std::ops::Deref;
use std::path::{Path, PathBuf};
use std::collections::{VecDeque, HashMap, BTreeMap};
use std::io::{Error, Result as IOResult, ErrorKind};
use std::sync::{Arc,
                OnceLock,
                atomic::{AtomicBool, AtomicU64, Ordering}};

use futures::{future::{FutureExt, BoxFuture}, stream::BoxStream, StreamExt};
use crossbeam_channel::bounded;
use async_lock::{Mutex, RwLock};
use async_channel::{Sender, Receiver, unbounded};
use dashmap::DashMap;
use lazy_static::lazy_static;
use bytes::BufMut;
use log::{info, error};
#[cfg(target_os = "linux")]
use libc::malloc_trim;
#[cfg(feature = "trace")]
use opentelemetry::{global,
                    metrics::Meter,
                    KeyValue};
#[cfg(feature = "trace")]
use pi_logger;
use pi_atom::Atom;
use pi_bon::{WriteBuffer, ReadBuffer, Encode, Decode, ReadBonErr};
use pi_guid::Guid;
use pi_async_rt::{lock::spin_lock::SpinLock,
                  rt::{AsyncRuntime, AsyncValue,
                       multi_thread::MultiTaskRuntime}};
use pi_async_transaction::{AsyncTransaction, Transaction2Pc, Transaction2PcAllConflicts, UnitTransaction, SequenceTransaction, TransactionTree, AsyncCommitLog, ErrorLevel, TransactionError,
                           manager_2pc::{Transaction2PcStatus, Transaction2PcManager}};
use pi_async_file::file::create_dir;
use pi_hash::XHashMap;

use crate::{Binary,
            KVAction,
            KVDBTableType,
            KVTableMeta,
            TableTrQos,
            KVDBCommitConfirm,
            KVTableTrError,
            MAX_TABLE_NAME_BYTES,
            TableKey,
            TableKeyVersion,
            Version,
            key_version::{KeyVersionConfig,
                          KeyVersionRegistry,
                          KeyVersions,
                          PrepareMode,
                          TableVersionContext,
            VersionReceipt},
            tables::{KVTable,
                     TableKV,
                     meta_table::{MetaTable,
                                  MetaTabTr},
                     mem_ord_table::{MemoryOrderedTable,
                                     MemOrdTabTr},
                     log_ord_table::{LogOrderedTable,
                                     LogOrdTabTr},
                     log_write_table::{LogWriteTable,
                                       LogWTabTr},
                     b_tree_ord_table::{DEFAULT_CACHE_SIZE, BtreeOrderedTable,
                                        BtreeOrdTabTr}},
            utils::{CreateTableOptions, KVDBEvent}};

///
/// 默认的数据库表元信息目录名
///
pub(crate) const DEFAULT_DB_TABLES_META_DIR: &str = ".tables_meta";

///
/// 默认的数据库表所在目录名
///
const DEFAULT_DB_TABLES_DIR: &str = ".tables";

///
/// 数据库未启动状态
///
const DB_UNSTARTUP_STATUS: u64 = 0;

///
/// 数据库正在初始化状态
///
const DB_INITING_STATUS: u64 = 1;

///
/// 数据库已初始化状态
///
const DB_INITED_STATUS: u64 = 2;

///
/// 数据库正在关闭状态
///
const DB_CLOSEING_STATUS: u64 = 3;

///
/// 数据库已关闭状态
///
const DB_CLOSED_STATUS: u64 = 4;

///
/// 启动数据库的源
///
const STARTUP_DB_SOURCE: &str = "Startup db";

///
/// 修复数据库时的源
///
const REPAIR_DB_SOURCE: &str = "Repair db";

const DEFAULT_KEY_VERSION_TTL: Duration = Duration::from_secs(60 * 60);
const DEFAULT_KEY_VERSION_TTL_POLL_INTERVAL: Duration = Duration::from_secs(3 * 60);

// 表缓存大小仪表
#[cfg(feature = "trace")]
static TABLE_CACHE_SIZE_METER: OnceLock<Meter> = OnceLock::new();

// 获取表缓存大小仪表
#[cfg(feature = "trace")]
pub(crate) async fn get_table_cache_size_meter<'a, R>(rt: R) -> &'a Meter
    where R: AsyncRuntime
{
    while !pi_logger::opentelemetry::is_init() {
        //跟踪系统还未初始化，则稍候重试
        rt.timeout(10000).await;
    }

    TABLE_CACHE_SIZE_METER.get_or_init(|| global::meter("table_cache_size"))
}

/// 键值对数据库管理器的一次性构建器。
///
/// 构建器拥有多线程 runtime、[`Transaction2PcManager`] 和三个词法路径；[`Self::new`] 不做
/// I/O，真正的目录创建、表加载、根 WAL 修复和后台任务启动发生在 [`Self::startup`] 或
/// [`Self::startup_with_listener`]。启动方法消费 `self`，同一构建器不能重复启动。
///
/// 该类型本身不提供配置持久化：LogOrdered/Btree 首次创建时的运行参数不会写入 Meta，重启
/// 使用固定默认值。启动不是事务原子操作，失败前已经创建的目录、文件或后台资源可能保留。
/// 真实入口和边界由 `tests/manager_contract.rs` 及恢复专项验证。
pub struct KVDBManagerBuilder<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    rt:                 MultiTaskRuntime<()>,               //异步运行时
    tr_mgr:             Transaction2PcManager<C, Log>,      //事务管理器
    db_path:            PathBuf,                            //数据库的表文件所在目录
    tables_meta_path:   PathBuf,                            //数据库的元信息表文件所在目录
    tables_path:        PathBuf,                            //数据库表文件所在目录
    key_version_ttl:    Duration,                           //Key 版本记录的生存时长
    key_version_ttl_poll_interval: Duration,                //Key 版本 TTL 固定轮询间隔
}

/*
* 键值对数据库管理器构建器同步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> KVDBManagerBuilder<C, Log> {
    /// 保存 runtime、事务管理器和数据库根路径，构造数据库启动计划。
    ///
    /// `rt` 必须是可继续接收任务的多线程 runtime；启动、表 collector、事件监听和最终 WAL
    /// 确认都会克隆并长期使用它。`tr_mgr` 的 commit logger 决定根 WAL 的真实位置，它可以与
    /// `path` 分离。`path` 只按 [`AsRef<Path>`] 复制，不会 canonicalize、创建、校验权限或
    /// 检查路径逃逸；相对路径仍相对于后续进程工作目录解释。
    ///
    /// 本函数为 O(p) 时间和空间，p 为路径长度；除分配和所有权转移外无副作用，不持锁、
    /// 不执行 I/O，也不会启动任务。
    pub fn new<P: AsRef<Path>>(rt: MultiTaskRuntime<()>,
                               tr_mgr: Transaction2PcManager<C, Log>,
                               path: P) -> Self {
        let db_path = path.as_ref().to_path_buf();
        let tables_meta_path = db_path.join(DEFAULT_DB_TABLES_META_DIR);
        let tables_path = db_path.join(DEFAULT_DB_TABLES_DIR);

        KVDBManagerBuilder {
            rt,
            tr_mgr,
            db_path,
            tables_meta_path,
            tables_path,
            key_version_ttl: DEFAULT_KEY_VERSION_TTL,
            key_version_ttl_poll_interval: DEFAULT_KEY_VERSION_TTL_POLL_INTERVAL,
        }
    }

    /// 设置 Key 版本记录的 TTL。
    ///
    /// 默认一小时；`Duration::ZERO` 关闭自动淘汰且不会创建后台 TTL 任务。配置只影响版本
    /// 证据的内存生命周期，不删除表数据，也不改变 WAL 和数据文件持久化语义。
    /// 最小单位是 1ms：非零 sub-ms 值按 1ms 处理，其余不足 1ms 的小数直接忽略。历史
    /// `BUG-KV-TTL-001` 已通过 deadline 基准向上取整修复；版本不得早于量化后的有效 TTL
    /// 淘汰，证据和边界见 `docs/KEY_VERSION_TTL_EARLY_EXPIRY_BUG.md`。
    pub fn key_version_ttl(mut self, ttl: Duration) -> Self {
        self.key_version_ttl = ttl;
        self
    }

    /// 设置 Key 版本 TTL 的固定轮询间隔。
    ///
    /// 默认三分钟。TTL 开启时 ZERO 属于非法启动配置；TTL 关闭时该值不生效。
    /// 间隔同样使用 1ms 最小单位并忽略其余小数；这一量化属于配置语义，不是
    /// `BUG-KV-TTL-001`。
    pub fn key_version_ttl_poll_interval(mut self, interval: Duration) -> Self {
        self.key_version_ttl_poll_interval = interval;
        self
    }
}

/*
* 键值对数据库管理器构建器异步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> KVDBManagerBuilder<C, Log> {
    /// 启动不带事件回调的数据库。
    ///
    /// `enable_accelerated_repair` 只传给 Btree 打开/恢复相关路径；无论它为 `false` 还是
    /// `true`，启动都会执行完整的内部 `try_repair` 根 WAL 扫描，而不是切换成另一套
    /// `try_quick_repair` 前导日志协议。该开关不改变 Meta、Memory、LogOrdered 或 LogWrite
    /// 的逻辑数据语义。
    ///
    /// 成功返回共享的 [`KVDBManager`]。未安装 listener，因此
    /// [`KVDBManager::report_transaction_info`] 会返回 `ConnectionAborted`。错误保留底层目录、
    /// 表加载或恢复原因，但启动不具备 rollback/cancellation 原子性，失败前已产生的文件系统
    /// 副作用不会由本 API 撤销。
    ///
    /// 运行时间为 O(t + w)，t 为 Meta 中用户表数，w 为待扫描/重放 WAL 字节数，并包含真实
    /// 异步文件 I/O、同步锁和表引擎打开成本。单 worker runtime 上的 repair 可能阻塞，见
    /// `FIND-REPAIR-001`。
    pub async fn startup(self, enable_accelerated_repair: bool) -> IOResult<KVDBManager<C, Log>> {
        self
            .startup_with_listener::<fn(&KVDBManager<C, Log>, &Transaction2PcManager<C, Log>, &mut Vec<KVDBEvent<Guid>>)>(enable_accelerated_repair, None)
            .await
    }

    /// 启动数据库，并可选安装批量事件回调。
    ///
    /// 启动顺序是：确保 Meta/用户表目录存在，打开内部 Meta 表，按 Meta 快照批量加载用户表，
    /// 调用内部 `try_repair` 修复未确认根 WAL，最后启动 listener 任务并把数据库标为可用。
    /// `enable_accelerated_repair` 的语义与 [`Self::startup`] 相同。
    ///
    /// `db_event_listener=None` 不创建事件通道。传入 `Some` 时，回调被保存到一个长期 runtime
    /// 任务中，并以 `FnMut(&KVDBManager, &Transaction2PcManager, &mut Vec<KVDBEvent>)` 形式同步、
    /// 串行调用。回调必须在返回前 `drain` 或 `clear` 已处理元素；框架不会清空 Vec。回调阻塞
    /// 会占用 worker，panic 会终止监听任务；通道无界，生产速度长期超过消费速度会增长内存。
    /// 回调可以读取共享 manager，但重入异步维护/DDL 时必须自行避免锁顺序问题。
    ///
    /// 返回、错误、副作用、取消安全和复杂度与 [`Self::startup`] 相同。启动过程会分配表注册表、
    /// channel 和表对象并执行文件 I/O；不是幂等的“探测”操作，也不保证多个进程或 manager
    /// 可以同时打开同一路径。真实 listener 契约由 `tests/manager_contract.rs` 验证。
    pub async fn startup_with_listener<F>(
        self,
        enable_accelerated_repair: bool,
        db_event_listener: Option<F>
    ) -> IOResult<KVDBManager<C, Log>>
    where F: FnMut(&KVDBManager<C, Log>, &Transaction2PcManager<C, Log>, &mut Vec<KVDBEvent<Guid>>) + Send + Sync + 'static
    {
        // TTL 配置必须先于目录、文件、channel 和表对象创建完成校验，保证 InvalidInput 零副作用。
        let key_version_config = KeyVersionConfig::new(self.key_version_ttl,
                                                       self.key_version_ttl_poll_interval)?;
        if !self.tables_meta_path.exists() {
            //指定路径的元信息表目录不存在，则创建
            let _ = create_dir(self.rt.clone(), self.tables_meta_path.clone()).await?;
        }

        if !self.tables_path.exists() {
            //指定路径的表目录不存在，则创建
            let _ = create_dir(self.rt.clone(), self.tables_path.clone()).await?;
        }

        //创建键值对数据库管理器
        let rt = self.rt;
        let tr_mgr = self.tr_mgr;
        let db_path = self.db_path;
        let tables_meta_path = self.tables_meta_path;
        let tables_path = self.tables_path;
        let tables = Arc::new(RwLock::new(XHashMap::default()));
        let (key_versions, key_version_ttl_receiver) = KeyVersionRegistry::new(key_version_config);
        let status = AtomicU64::new(DB_INITING_STATUS);
        let (notifier, listener) = if db_event_listener.is_some() {
            let (notifier, listener) = unbounded();
            (Some(notifier), Some(listener))
        } else {
            (None, None)
        };
        let inner = InnerKVDBManager {
            rt,
            tr_mgr,
            db_path,
            tables_meta_path,
            tables_path,
            tables,
            key_versions,
            status,
            listener,
            notifier,
        };
        let db_mgr = KVDBManager(Arc::new(inner));

        //加载并注册元信息表
        let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        let meta_table: MetaTable<C, Log> =
            MetaTable::new(db_mgr.0.rt.clone(),
                           db_mgr.tables_meta_path().to_path_buf(),
                           meta_table_name.clone(),
                           512 * 1024 * 1024,
                           2 * 1024 * 1024,
                           None,
                           2 * 1024 * 1024,
                           true,
                           16 * 1024 * 1024,
                           60 * 1000,
                           db_mgr.0.notifier.clone()).await;
        let meta_versions = db_mgr.0.key_versions.create_table_versions();
        db_mgr.0.key_versions.install(meta_table_name.clone(), meta_versions.clone());
        db_mgr.0.tables.write().await.insert(meta_table_name.clone(),
                                             RegisteredTable::new(KVDBTable::MetaTab(meta_table),
                                                                  meta_versions));

        //根据元信息表的元信息，加载其它表，加载操作使用的事务，不需要预提交和提交
        let mut tr = db_mgr
            .transaction(Atom::from(STARTUP_DB_SOURCE),
                         true,
                         1000,
                         1000)
            .unwrap();
        let mut meta_iterator = tr
            .values(meta_table_name.clone(),
                    None,
                    false)
            .await
            .unwrap();
        let mut table_metas_buf = Vec::with_capacity(8192);
        let default_log_table_options = CreateTableOptions::LogOrdTab(512 * 1024 * 1024,
                                                                      2 * 1024 * 1024,
                                                                      2 * 1024 * 1024);
        let default_b_tree_table_options = CreateTableOptions::BtreeOrdTab(16 * 1024 * 1024,
                                                                           true);
        let now = Instant::now();
        while let Some((key, value)) = meta_iterator.next().await {
            let table_name = match binary_to_table(&key) {
                Err(e) => {
                    //反序列化表名失败
                    return Err(Error::new(ErrorKind::Other, format!("From binary to table name failed, reason: {:?}", e)));
                },
                Ok(table_name) => {
                    //反序列化表名成功
                    table_name
                }
            };

            validate_table_name(&table_name,
                                ErrorKind::InvalidData,
                                "load table metadata")?;

            if table_name == meta_table_name {
                //忽略元信息表
                continue;
            }
            let table_meta = KVTableMeta::from(value);

            if table_metas_buf.len() < 8192 {
                //填充表元信息，并继续迭代下一个表元信息
                match table_meta.table_type() {
                    KVDBTableType::LogOrdTab => {
                        table_metas_buf.push((table_name, table_meta, Some(default_log_table_options.clone())));
                    },
                    KVDBTableType::BtreeOrdTab => {
                        table_metas_buf.push((table_name, table_meta, Some(default_b_tree_table_options.clone())));
                    },
                    _ => {
                        table_metas_buf.push((table_name, table_meta, None));
                    },
                }
                continue;
            }
            let mut table_metas = Vec::with_capacity(table_metas_buf.len());
            swap(&mut table_metas_buf, &mut table_metas);

            //异步批量加载表
            if let Err(e) = tr.create_multiple_tables(
                table_metas,
                true,
                enable_accelerated_repair
            ).await {
                //加载指定的表失败，则立即返回错误原因
                db_mgr.0.status.store(DB_UNSTARTUP_STATUS, Ordering::SeqCst);
                return Err(Error::new(ErrorKind::Other,
                                      format!("Load table failed, tables_path: {:?}, reason: {:?}",
                                              db_mgr.tables_path(),
                                              e)));
            }
        }
        if table_metas_buf.len() > 0 {
            //异步批量加载剩余的表
            if let Err(e) = tr.create_multiple_tables(
                table_metas_buf,
                true,
                enable_accelerated_repair
            ).await {
                //加载指定的表失败，则立即返回错误原因
                db_mgr.0.status.store(DB_UNSTARTUP_STATUS, Ordering::SeqCst);
                return Err(Error::new(ErrorKind::Other,
                                      format!("Load table failed, tables_path: {:?}, reason: {:?}",
                                              db_mgr.tables_path(),
                                              e)));
            }
        }
        info!("Load db succeeded, tables: {:?}, time: {:?}",
            db_mgr.table_size().await,
            now.elapsed());
        drop(meta_iterator);
        drop(tr);

        //如果有未确认的提交日志，则尝试修复数据库表数据
        let now = Instant::now();
        match db_mgr.try_repair(enable_accelerated_repair).await {
            Err(e) => {
                //有未确认的提交日志，且尝试修复数据库表数据失败，则立即返回错误原因
                return Err(e);
            },
            Ok((repaired_log_len, repaired_bytes_len)) => {
                //尝试修复数据库表数据成功
                if repaired_log_len > 0 {
                    //未确认的提交日志
                    info!("Repair db succeeded, logs: {}, bytes: {}, time: {:?}",
                        repaired_log_len,
                        repaired_bytes_len,
                        now.elapsed());
                }
            }
        }

        // repair 期间产生的版本只服务内部恢复，不得暴露给启动后的外部缓存。
        db_mgr.0.key_versions.clear_records();
        db_mgr.0.key_versions.start_ttl_task(db_mgr.0.rt.clone(), key_version_ttl_receiver);

        if let Some(mut handle) = db_event_listener {
            //指定了数据库事件监听器
            let db_mgr_copy = db_mgr.clone();
            let listener = db_mgr
                .0
                .listener
                .as_ref()
                .unwrap()
                .clone();


            let _ = db_mgr.0.rt.spawn(async move {
                let mut events = Vec::with_capacity(3072);
                loop {
                    let mut try_count = 5usize;
                    while events.len() < 3072 {
                        //未达单次事件处理上限
                        if let Ok(event) = listener.try_recv() {
                            //有事件
                            try_count = 5; //重置重试次数
                            events.push(event);
                        } else {
                            //无事件
                            if try_count > 0 {
                                //未达重试限制，则稍候重试
                                try_count = try_count.checked_sub(1).unwrap_or(0);
                                db_mgr_copy
                                    .0
                                    .rt
                                    .timeout(10)
                                    .await;
                                continue;
                            }

                            //已达重试限制
                            if events.len() > 0 {
                                //有待处理的事件，则立即处理
                                try_count = 5; //重置重试次数
                                handle(&db_mgr_copy, &db_mgr_copy.0.tr_mgr, &mut events);
                                break;
                            } else {
                                //没有待处理的事件，则等待
                                if let Ok(event) = listener.recv().await {
                                    //有事件
                                    try_count = 5; //重置重试次数
                                    events.push(event);
                                }
                            }
                        }
                    }

                    if events.len() > 0 {
                        //有待处理的事件，则立即处理
                        handle(&db_mgr_copy, &db_mgr_copy.0.tr_mgr, &mut events);
                    }
                }
            });
        }

        //启动跟踪系统
        #[cfg(feature = "trace")]
        {
            let rt_copy = db_mgr.0.rt.clone();
            let db_mgr_copy = db_mgr.clone();
            let _ = db_mgr.0.rt.spawn(async move {
                loop_tracing(rt_copy, db_mgr_copy, 15000).await;
            });
        }

        db_mgr.0.status.store(DB_INITED_STATUS, Ordering::SeqCst); //设置数据库状态为已初始化
        info!("Startup db succeeded");

        //在Linux下启动完成后清理一次内存
        #[cfg(target_os = "linux")]
        db_mgr.cleanup_buffer_after_collect_table();

        Ok(db_mgr)
    }
}

/// 已启动数据库的共享管理句柄。
///
/// clone 只增加内部 [`Arc`] 强引用，所有 clone 共享表注册表、关闭状态、事务管理器、根 WAL、
/// listener 通道和 runtime。句柄被 drop 不等于数据库关闭：表 collector、listener 和 runtime
/// 任务可能继续持有 manager/table 引用；[`Self::close`] 也只是立即禁止新事务的软关闭标记，
/// 不等待数据文件、WAL 确认、后台任务或文件句柄释放。完整边界见
/// `docs/SEMANTIC_CONTRACTS.md#q-close-001` 和 `FIND-LIFE-001`。
///
/// 管理器允许跨线程共享；各表注册变更通过异步 `RwLock`，状态通过原子变量，事务和 logger
/// 的并发安全依赖各自公开契约。真实管理 API 矩阵见 `tests/manager_contract.rs`。
pub struct KVDBManager<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
>(Arc<InnerKVDBManager<C, Log>>);

// SAFETY: manager 只移动一个 Arc。Inner 中 registry 使用 async RwLock，状态使用 AtomicU64，
// channel/runtime/Transaction2PcManager/AsyncCommitLog 均由其类型契约提供跨线程同步；路径在
// 构造后只读。C 不以裸值直接存入 manager，而只出现在已要求 Send 的事务依赖类型中。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Send for KVDBManager<C, Log> {}
// SAFETY: 所有通过共享 manager 访问的可变状态都位于上述同步原语或依赖类型之后；公开路径
// 借用只读，Arc clone/drop 使用原子引用计数。该 impl 不赋予 listener 回调重入安全，调用方
// 仍须遵守 FIND-ASYNC-001 记录的锁顺序边界。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Sync for KVDBManager<C, Log> {}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Clone for KVDBManager<C, Log> {
    fn clone(&self) -> Self {
        KVDBManager(self.0.clone())
    }
}

/*
* 键值对数据库管理器同步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>
> KVDBManager<C, Log> {
    /// 借用构建器保存的数据库根路径。
    ///
    /// 返回值与 `self` 生命周期相同，是构建时路径的词法副本，不保证 canonical、存在、可写，
    /// 也不表示 commit logger 的 WAL 路径。O(1)、纯只读、无分配、无锁和无 I/O；所有 clone
    /// 返回相同路径内容。
    pub fn db_path(&self) -> &Path {
        &self.0.db_path
    }

    /// 借用内部 Meta 表目录。
    ///
    /// 路径固定为 `db_path/.tables_meta`，启动成功时应已存在；它是 Meta 表数据目录，不是
    /// 根 WAL 目录。返回借用不转移所有权，O(1)、无锁、无分配和无 I/O。
    pub fn tables_meta_path(&self) -> &Path {
        &self.0.tables_meta_path
    }

    /// 借用持久化用户表的父目录。
    ///
    /// 路径固定为 `db_path/.tables`。Memory 表即使 `persistence=true` 也没有该目录下的数据
    /// 文件；该标志只使动作进入根 WAL。返回借用为 O(1)、只读、无锁、无分配和无 I/O。
    pub fn tables_path(&self) -> &Path {
        &self.0.tables_path
    }

    /// 创建尚未注册到两阶段管理器的根事务句柄。
    ///
    /// `source` 被事务持有并用于 UID/统计/事件；`is_writable` 决定 prepare/commit 主路径，
    /// 但当前动作 API 没有统一拒绝只读写入，只读写可能先返回成功、随后在 prepare 快路中
    /// 被静默丢弃。这是 `FIND-TR-001` 的当前实现事实，不是最终或最佳只读契约。
    /// `prepare_timeout` 和 `commit_timeout` 仅保存并传给子事务，当前没有计时、取消或错误路径，
    /// 见 `FIND-TIMEOUT-001`；`0` 也不会被本函数拒绝。
    ///
    /// 数据库状态为初始化中或已初始化时返回 `Some(RootTr)`；调用 [`Self::close`] 后返回
    /// `None`。创建本身不会分配事务 UID、注册 active transaction 或写 WAL，这些发生在
    /// [`KVDBTransaction::prepare_modified`]。在 close 前已经创建的句柄当前仍可继续 prepare/
    /// commit，这是软关闭边界而不是 graceful shutdown 保证。
    ///
    /// 根事务是否需要持久化由实际触达且需要持久化的写子事务决定。函数为摊销 O(1)，会分配
    /// Arc 和空的子事务 map/list，不执行文件 I/O、不 await；句柄反向持有 manager clone，因而
    /// 会延长数据库对象生命周期，但没有从 manager 指回该事务的环，直到 prepare 注册为止。
    pub fn transaction(&self,
                       source: Atom,
                       is_writable: bool,
                       prepare_timeout: u64,
                       commit_timeout: u64) -> Option<KVDBTransaction<C, Log>> {
        let status = self.0.status.load(Ordering::Relaxed);
        if status != DB_INITING_STATUS && status != DB_INITED_STATUS {
            //当前数据库状态不允许创建键值对数据库的根事务，则立即返回空
            return None;
        }

        let tid = SpinLock::new(None);
        let cid = SpinLock::new(None);
        let status = SpinLock::new(Transaction2PcStatus::Start);
        let childs_map = SpinLock::new(XHashMap::default());
        let childs = SpinLock::new(KVDBChildTrList::new());
        let db_mgr = self.clone();

        let inner = InnerRootTransaction {
            source,
            tid,
            cid,
            status,
            writable: is_writable,
            persistence: AtomicBool::new(false), //默认键值对数据库的根事务不持久化
            prepare_timeout,
            commit_timeout,
            childs_map,
            childs,
            db_mgr,
            version_context: SpinLock::new(None),
        };

        Some(KVDBTransaction::RootTr(RootTransaction(Arc::new(inner))))
    }

    /// 请求当前进程的 glibc allocator 归还可释放页。
    ///
    /// 仅 Linux 提供，直接调用 `malloc_trim(0)`；`true` 表示 allocator 报告释放了内存，
    /// `false` 不表示没有可回收对象。该操作作用于整个进程而非单个数据库，可能同步扫描
    /// allocator 并阻塞调用线程，非纯函数、非确定性，也不保证降低 RSS。
    ///
    /// # Safety
    ///
    /// FFI 参数 `0` 是 glibc 允许的 pad 值，调用不传入 Rust 指针。进程所用 allocator 必须
    /// 与链接到的 libc 实现兼容；本 API 不应在时延敏感热路径频繁调用。
    #[cfg(target_os = "linux")]
    pub fn cleanup_buffer_after_collect_table(&self) -> bool {
        match unsafe { malloc_trim(0) } {
            0 => false,
            _ => true,
        }
    }

    /// 设置共享 manager 的软关闭状态并立即禁止创建新事务。
    ///
    /// 若两阶段管理器当前没有已注册事务，状态直接设为 Closed；否则设为 Closing。所有 clone
    /// 立即观察同一关闭结果，重复调用不会重新开放数据库。调用无返回值，不等待活跃事务、
    /// 子表异步持久化、根 WAL 确认、collector/listener 退出或文件句柄释放，也不取消 close 前
    /// 已创建但尚未 prepare 的事务句柄。最后一个 active transaction 完成后当前没有自动把
    /// Closing 推进为 Closed；再次调用本方法才会重写状态。
    ///
    /// 这是 `Q-CLOSE-001` / `FIND-CLOSE-001` 记录的当前实现，不是最终或最佳 shutdown API。
    /// 操作为 O(1)，读取事务 registry 长度并原子写状态，不执行文件 I/O、不 await、不调用
    /// 用户回调。真实边界由 `tests/manager_contract.rs` 验证。
    pub fn close(&self) {
        if self.0.tr_mgr.transaction_len() == 0 {
            //如果当前事务管理器没有任何正在执行的事务，则设置数据库状态为已关闭
            self.0.status.store(DB_CLOSED_STATUS, Ordering::SeqCst);
        } else {
            //如果当前事务管理器还有任何正在执行的事务，则设置数据库状态为正在状态
            self.0.status.store(DB_CLOSEING_STATUS, Ordering::SeqCst);
        }
    }
}

/*
* 键值对数据库管理器异步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>
> KVDBManager<C, Log> {
    /// 读取一个 Key 的当前逻辑值及同一 publication 窗口中的公开版本。
    ///
    /// 该入口不创建事务、不获取 prepare 锁，也不写 WAL 或数据文件。版本命中时只读取记录；
    /// 缺席时会分配独立 Guid 并登记首次观察和 TTL，因此它不是纯函数。LogWrite 的逻辑值固定
    /// 为 `None`；Btree 的真实点读错误返回可恢复 `Common(Normal)`，绝不伪装成 Key 不存在。
    ///
    /// 表名长度须为 1..=4096 字节，Key 长度须为 1..=u16::MAX。返回版本必须与值成对缓存，
    /// 并且只能进入 `prepare_with_version -> commit_with_version` 独立协议；禁止与普通事务 API
    /// 混用。完整契约见 `docs/KEY_VERSION_PUBLICATION_DESIGN.md#key-version-public-api` 和
    /// `docs/PI_DB_SERVER_KEY_VERSION_API_HANDOFF.md#pi-db-server-query-with-version`。
    pub async fn query_with_version(&self,
                                    table: Atom,
                                    key: Binary)
        -> Result<(Option<Binary>, Version), KVTableTrError> {
        validate_version_table_key(&table, &key, "query with version")?;
        let registered = {
            self.0
                .tables
                .read()
                .await
                .get(&table)
                .cloned()
        }.ok_or_else(|| {
            KVTableTrError::new_transaction_error(
                ErrorLevel::Normal,
                format!("Query with version failed, table: {:?}, reason: table not found",
                        table.as_str()))
        })?;

        let _publication = registered.versions.publication().read().await;
        let value = match &registered.table {
            KVDBTable::MetaTab(table) => table.query_committed(&key),
            KVDBTable::MemOrdTab(table) => table.query_committed(&key),
            KVDBTable::LogOrdTab(table) => table.query_committed(&key),
            KVDBTable::LogWTab(_) => None,
            KVDBTable::BtreeOrdTab(table) => table
                .query_committed(&key)
                .map_err(|e| KVTableTrError::new_transaction_error(ErrorLevel::Normal, e))?,
        };
        let version = registered
            .versions
            .first_observation(key,
                               value.is_some(),
                               || self.0.tr_mgr.alloc_transaction_uid());

        Ok((value, version))
    }

    /// 克隆指定名称的内部表句柄。
    ///
    /// 这是 crate 内模块间接口；`table_name` 只在 await 期间借用。存在时返回共享表 clone，
    /// 不存在或已经从 registry 移除时返回 `None`。返回句柄可继续延长表生命周期，因此 registry
    /// 删除不等于立即释放文件、collector 或内存。平均 O(1)，短暂获取异步 registry 读锁，
    /// 不执行表 I/O。
    pub(crate) async fn get_table(&self, table_name: &Atom) -> Option<KVDBTable<C, Log>> {
        if let Some(table) = self.0.tables.read().await.get(table_name) {
            Some(table.table.clone())
        } else {
            None
        }
    }

    /// 判断表名当前是否注册。
    ///
    /// `table_name` 不会被保存；内部 `.tables_meta` 也计为存在。结果是调用时 registry 状态，
    /// 不是事务快照，返回后可立即因 DDL 改变。平均 O(1)，获取一次异步读锁，无文件 I/O。
    pub async fn is_exist(&self, table_name: &Atom) -> bool {
        self.0.tables.read().await.contains_key(table_name)
    }

    /// 返回当前 registry 条目数。
    ///
    /// 数量包含内部 `.tables_meta`，不只包含用户表；是瞬时值而非跨调用稳定快照。平均 O(1)，
    /// 获取一次异步读锁，无分配和文件 I/O。
    pub async fn table_size(&self) -> usize {
        self.0.tables.read().await.len()
    }

    /// 克隆当前 registry 中全部表名。
    ///
    /// 返回列表包含内部 `.tables_meta`，顺序来自 hash map，未排序且不稳定。每个 [`Atom`] clone
    /// 保持名称有效，但不保持表仍注册；调用方需要稳定顺序时必须自行排序。O(n) 时间和 O(n)
    /// 新空间，迭代期间持异步 registry 读锁，不执行文件 I/O。
    pub async fn tables(&self) -> Vec<Atom> {
        let mut table_names = Vec::new();
        for key in self.0.tables.read().await.keys() {
            table_names.push(key.clone());
        }

        table_names
    }

    /// 返回指定表实现报告的数据位置词法副本。
    ///
    /// 缺表返回 `None`。Meta、LogOrdered 和 LogWrite 返回各自日志目录；Btree 返回 redb 数据
    /// 文件 `tables_path/<table>/table.dat`，不是其父目录；Memory 无论 `persistence` 为何都返回
    /// `None`，因为它没有存储引擎数据文件。`None` 因此不能单独区分“缺表”和“已存在
    /// Memory 表”，应与 [`Self::is_exist`] 联合判断。调用方不得把不同表 variant 的返回路径
    /// 统一当作目录执行操作。
    ///
    /// 返回路径不保证 canonical、当前存在或可访问。平均 O(1) 查找加 O(p) 路径复制，短暂持
    /// registry 读锁，不访问文件系统。
    pub async fn table_path(&self, table_name: &Atom) -> Option<PathBuf> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => None,
            Some(KVDBTable::MetaTab(table)) => {
                if let Some(path) = table.path() {
                    Some(path.to_path_buf())
                } else {
                    None
                }
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                if let Some(path) = table.path() {
                    Some(path.to_path_buf())
                } else {
                    None
                }
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                if let Some(path) = table.path() {
                    Some(path.to_path_buf())
                } else {
                    None
                }
            },
            Some(KVDBTable::LogWTab(table)) => {
                if let Some(path) = table.path() {
                    Some(path.to_path_buf())
                } else {
                    None
                }
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                if let Some(path) = table.path() {
                    Some(path.to_path_buf())
                } else {
                    None
                }
            },
        }
    }

    /// 查询表的持久化标志。
    ///
    /// 缺表返回 `None`；存在时返回 `Some`。Meta、LogOrdered、LogWrite、Btree 当前总为
    /// `Some(true)`；Memory 返回建表元信息中的标志。Memory 的 `true` 仅表示动作进入根 WAL，
    /// 不会创建数据文件。结果是表实例属性，不表示当前事务已有待持久化动作或最终确认完成。
    /// 平均 O(1)，短暂持 registry 读锁，无分配和文件 I/O。
    pub async fn is_persistent_table(&self, table_name: &Atom) -> Option<bool> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => None,
            Some(KVDBTable::MetaTab(table)) => {
                Some(table.is_persistent())
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                Some(table.is_persistent())
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                Some(table.is_persistent())
            },
            Some(KVDBTable::LogWTab(table)) => {
                Some(table.is_persistent())
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                Some(table.is_persistent())
            },
        }
    }

    /// 查询表是否报告为有序表。
    ///
    /// 缺表返回 `None`。当前五种内部/用户表都返回 `Some(true)`，包括只写语义的 LogWrite；
    /// 该布尔值不能推导某表支持 query/delete/stream 的完整能力。平均 O(1)，短暂持 registry
    /// 读锁，无分配和文件 I/O。
    pub async fn is_ordered_table(&self, table_name: &Atom) -> Option<bool> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => None,
            Some(KVDBTable::MetaTab(table)) => {
                Some(table.is_ordered())
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                Some(table.is_ordered())
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                Some(table.is_ordered())
            },
            Some(KVDBTable::LogWTab(table)) => {
                Some(table.is_ordered())
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                Some(table.is_ordered())
            },
        }
    }

    /// 获取表实现当前报告的记录数。
    ///
    /// 缺表返回 `None`。Meta/Memory/LogOrdered/LogWrite 从当前内存根读取；Btree 同步打开 redb
    /// read transaction，再把只写 cache 粗略叠加到持久基线。Btree 当前会把 begin_read、
    /// open_table 或 `table.len()` 错误折叠为 `Some(0)`，且 tombstone/overlay 计数语义仍有
    /// `FIND-TABLE-002` 风险；因此该值不是可用于诊断存储健康的无损 Result。
    ///
    /// 调用持 registry 异步读锁并进入表内同步锁；Btree 还可能在 runtime worker 上执行同步
    /// 文件支持读取，复杂度约 O(c log d)，c 为 cache key 数、d 为 redb 记录数。其它表通常
    /// 为 O(1)。结果是调用时观察值，不与外部事务建立原子快照。
    pub async fn table_record_size(&self, table_name: &Atom) -> Option<usize> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => None,
            Some(KVDBTable::MetaTab(table)) => {
                Some(table.len())
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                Some(table.len())
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                Some(table.len())
            },
            Some(KVDBTable::LogWTab(table)) => {
                Some(table.len())
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                Some(table.len())
            },
        }
    }

    /// 获取表实现当前报告的内存缓存 payload 字节数。
    ///
    /// 缺表返回 `None`；存在空表通常返回 `Some(0)`。该值由表内 COW root/cache 的
    /// `full_bytes_size` 提供，不包含 Arc、树节点、allocator、文件缓存、redb 页面缓存、WAL
    /// 文件或后台任务开销，不能当作进程 RSS。Btree 只统计只写 cache，不统计 redb 数据。
    ///
    /// 调用短暂持 registry 异步读锁和表内同步锁，通常 O(1)、无文件 I/O；并发写入后返回值
    /// 只代表本次读取时刻，不是事务快照。
    pub async fn table_cache_size(&self, table_name: &Atom) -> Option<u64> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => None,
            Some(KVDBTable::MetaTab(table)) => {
                Some(table.size())
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                Some(table.size())
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                Some(table.size())
            },
            Some(KVDBTable::LogWTab(table)) => {
                Some(table.size())
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                Some(table.size())
            },
        }
    }

    /// 强制根 commit logger 轮换到一个新的 checkpoint。
    ///
    /// 成功返回 `pi_store::LogFile::split` 刚创建的可写 WAL 文件索引。依赖中的
    /// `current_check_point()` 读取的是内部 `log_id` 的下一个待分配值，因此本方法成功后该值
    /// 等于“返回索引 + 1”，不能把两者误认为同一个编号。每次调用都会创建/切换可写 WAL
    /// 文件并把前一 checkpoint 加入只读确认队列，即使当前没有事务；因此不是纯函数，也不是
    /// 幂等操作。它不会追加一条业务事务日志，`append_total_count` 不应因此增加。
    ///
    /// 该异步操作获取 logger checkpoint 锁并执行真实文件 I/O；并发 append/confirm/轮换按
    /// `pi_store::CommitLogger` 锁顺序串行。错误原样作为 `io::Error` 返回，已发生的部分文件
    /// 副作用不由本 API rollback。真实副作用由 `tests/manager_contract.rs` 验证。
    pub async fn append_new_commit_log(&self) -> IOResult<usize> {
        let commit_logger = self.0.tr_mgr.commit_logger();
        commit_logger.append_check_point().await
    }

    /// 调用表实现的整理准备阶段。
    ///
    /// `table_name` 不会被保存。缺表当前静默返回 `Ok(())`；Memory/Btree 的准备阶段也是 no-op，
    /// Meta/LogOrdered/LogWrite 会 split 各自日志文件。成功只表示表级准备返回成功，不自动调用
    /// [`Self::collect_table`]，也不建立只能由同一调用者消费的一次性 token。
    ///
    /// 当前实现从取得 registry 读 guard 到表 future 完成期间一直持有该 guard，DDL registry
    /// 写入可能被长 I/O 阻塞，见 `FIND-ASYNC-001`。底层表错误被格式化并统一包装为
    /// `io::ErrorKind::Other`，错误等级信息不会结构化保留。非幂等表可能轮换文件。
    pub async fn ready_collect_table(&self, table_name: &Atom) -> IOResult<()> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => (),
            Some(KVDBTable::MetaTab(table)) => {
                if let Err(e) = table.ready_collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                if let Err(e) = table.ready_collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                if let Err(e) = table.ready_collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::LogWTab(table)) => {
                if let Err(e) = table.ready_collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                if let Err(e) = table.ready_collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
        }

        Ok(())
    }

    /// 直接调用表实现的整理/压缩阶段。
    ///
    /// 缺表当前静默返回 `Ok(())`；Memory 为 no-op，日志表执行真实日志 collect，Btree 执行
    /// redb 持久化/compact 路径。API 不检查调用方是否先调用 [`Self::ready_collect_table`]，
    /// 不返回整理后的大小或记录数，也不保证与并发事务形成全库级原子边界。
    ///
    /// 调用可能长时间执行文件 I/O、同步 redb 锁和 runtime timeout；整个 await 期间当前仍持
    /// registry 读 guard，见 `FIND-ASYNC-001`。表错误统一降为 `io::ErrorKind::Other`；Btree
    /// collect 的重试控制另见 `FIND-TABLE-001`。该方法有文件副作用且不保证幂等。
    pub async fn collect_table(&self, table_name: &Atom) -> IOResult<()> {
        match self.0.tables.read().await.get(table_name).map(|registered| &registered.table) {
            None => (),
            Some(KVDBTable::MetaTab(table)) => {
                if let Err(e) = table.collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::MemOrdTab(table)) => {
                if let Err(e) = table.collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::LogOrdTab(table)) => {
                if let Err(e) = table.collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::LogWTab(table)) => {
                if let Err(e) = table.collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
            Some(KVDBTable::BtreeOrdTab(table)) => {
                if let Err(e) = table.collect().await {
                    return Err(Error::new(ErrorKind::Other, format!("{:?}", e)));
                }
            },
        }

        Ok(())
    }

    /// 向可选 listener 通道发送一个事务信息报告请求事件。
    ///
    /// 安装 listener 时，本方法把 [`KVDBEvent::ReportTrInfo`] 写入无界通道并返回；`Ok(())`
    /// 只证明 send 接受了事件，不证明 listener 已被调度、已经清空该批事件或成功产出报告。
    /// listener 回调可通过参数中的 [`Transaction2PcManager`] 自行读取当时统计。未安装 listener
    /// 时返回 `io::ErrorKind::ConnectionAborted`。
    ///
    /// 发送为异步安全且允许多线程并发，但通道无背压，持续调用可能增长内存。该 API 不持表
    /// registry 锁、不执行文件 I/O、不等待用户回调，也不保证事件相对 collector 通知的全局
    /// 顺序。真实投递由 `tests/manager_contract.rs` 验证。
    pub async fn report_transaction_info(&self) -> IOResult<()> {
        if let Some(notifier) = self.0.notifier.as_ref() {
            if let Err(e) = notifier.send(KVDBEvent::ReportTrInfo).await {
                Err(Error::new(ErrorKind::ConnectionAborted,
                               format!("Report transaction info failed, reason: {:?}", e)))
            } else {
                Ok(())
            }
        } else {
            Err(Error::new(ErrorKind::ConnectionAborted,
                           format!("Report transaction info failed, reason: invalid notifier")))
        }
    }

    // 尝试幂等的重播未确认的提交日志，并修复数据库表数据
    // 注意如果在只有单个线程的运行时修复或并发修复，则可能会发生阻塞
    pub(crate) async fn try_repair(&self, enable_accelerated_repair: bool) -> IOResult<(usize, usize)> {
        //构建重播回调
        let db_mgr = self.clone();
        // pi_store 当前会把 replay callback 的 io::Error 统一包装成 Other。该原子只标记
        // “删表持久化输入非法”这一冻结分支，使 try_repair 能在不修改存储依赖和其它恢复
        // 错误语义的前提下恢复 InvalidData 分类。
        let invalid_remove_data = Arc::new(AtomicBool::new(false));
        let invalid_remove_data_copy = invalid_remove_data.clone();

        let tables = Arc::new(Mutex::new(BTreeMap::new()));
        let tables_copy = tables.clone();
        let replay_callback = move |commit_uid: Guid, prepare_output: Vec<u8>| -> IOResult<()> {
            //异步执行重播
            let db_mgr_copy = db_mgr.clone();
            let commit_uid_copy = commit_uid.clone();
            let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
            let (sender, receiver) = bounded(1);

            let tables_clone = tables_copy.clone();
            let invalid_remove_data = invalid_remove_data_copy.clone();
            let boxed = async move {
                let bytes_len = prepare_output.len(); //获取日志缓冲区长度
                let mut offset = 0; //日志缓冲区偏移
                let bytes = prepare_output.as_slice();
                let uid = u128::from_le_bytes(bytes[0..16].try_into().unwrap()); //获取事务唯一id
                let transaciton_uid = Guid(uid);
                offset += 16; //移动缓冲区指针

                if let Some(tr) = db_mgr_copy.transaction(Atom::from(REPAIR_DB_SOURCE),
                                                          true,
                                                          5000,
                                                          5000)
                {
                    //创建数据库事务成功，则迭代日志缓冲区中，本次未确认的提交日志中执行写操作的表和相关键值对
                    //迭代完成后，则可以恢复本次未确认的提交日志对表的内存键值对的修改，并生成对应的键值对操作记录
                    while offset < bytes_len {
                        //获取表名、操作的键值对数量和新的日志缓冲区偏移
                        let (table, kvs_len, new_offset) =
                            <MetaTable<C, Log> as KVTable>::get_init_table_prepare_output(&prepare_output, offset);

                        //获取操作的表键值列表和新的日志缓冲区偏移
                        let (writes, new_offset)
                            = <MetaTable<C, Log> as KVTable>::get_all_key_value_from_table_prepare_output(&prepare_output, &table, kvs_len, new_offset);

                        if table == meta_table_name {
                            //未确认的提交日志操作的表是元信息表，则创建或删除表
                            for write in writes {
                                if let Some(value) = write.value {
                                    //有值，则创建表
                                    let table_name = match binary_to_table(&write.key) {
                                        Err(e) => {
                                            //反序列化表名失败
                                            panic!("From binary to table name failed, reason: {:?}", e);
                                        },
                                        Ok(table_name) => {
                                            //反序列化表名成功
                                            table_name
                                        }
                                    };
                                    let table_meta = KVTableMeta::from(value);

                                    if let Err(e) = tr.repair_create_table(
                                        table_name.clone(),
                                        table_meta.clone(),
                                        enable_accelerated_repair,
                                    ).await {
                                        //重播的创建表失败，则立即返回错误原因
                                        let _ = sender.send(Err(Error::new(ErrorKind::Other, format!("Repair tables meta failed, transaction_uid: {:?}, commit_uid: {:?}, table_name: {:?}, table_meta: {:?}, reason: {:?}", transaciton_uid, commit_uid_copy, table_name, table_meta, e))));
                                        return;
                                    }
                                } else {
                                    //无值，则删除表
                                    // Meta tombstone 的 Key 与建表记录相同，都是 table_to_binary
                                    // 生成的 BON Atom，不能把编码字节直接当 UTF-8 表名。该分支
                                    // 只服务删表 WAL 恢复；解码失败表示持久化日志损坏。
                                    // CONTRACT-DDL-REMOVE-001 / BUG-DDL-REMOVE-001：
                                    // docs/SEMANTIC_CONTRACTS.md#contract-ddl-remove-crash-durability。
                                    let table_name = match binary_to_table(&write.key) {
                                        Err(e) => {
                                            invalid_remove_data.store(true, Ordering::Release);
                                            let _ = sender.send(Err(Error::new(
                                                ErrorKind::InvalidData,
                                                format!("Repair removed table failed, transaction_uid: {:?}, commit_uid: {:?}, table_key_bytes: {}, reason: decode table name failed: {:?}",
                                                        transaciton_uid,
                                                        commit_uid_copy,
                                                        write.key.len(),
                                                        e),
                                            )));
                                            return;
                                        },
                                        Ok(table_name) => table_name,
                                    };

                                    if let Err(e) = tr.repair_remove_table(table_name.clone()).await {
                                        // 重播删表失败时保留底层 ErrorKind：保留 Meta 名称或非法
                                        // 持久化名称必须作为 InvalidData 传播，不能被改写为 Other。
                                        if e.kind() == ErrorKind::InvalidData {
                                            invalid_remove_data.store(true, Ordering::Release);
                                        }
                                        let _ = sender.send(Err(Error::new(
                                            e.kind(),
                                            format!("Repair tables meta failed, transaction_uid: {:?}, commit_uid: {:?}, table_name: {:?}, reason: {:?}",
                                                    transaciton_uid,
                                                    commit_uid_copy,
                                                    table_name,
                                                    e),
                                        )));
                                        return;
                                    }
                                }
                            }
                        } else {
                            //未确认的提交日志操作的表是其它表，则执行本次未确认的提交日志中指定表的键值对写操作
                            tables_clone
                                .lock()
                                .await
                                .insert(table.clone(), ());
                            for write in writes {
                                if write.exist_value() {
                                    //有值，则执行插入或更新操作
                                    if let Err(e) = tr.upsert(vec![write]).await {
                                        //重播的插入或更新操作失败，则立即返回错误原因
                                        let _ = sender.send(Err(Error::new(ErrorKind::Other, format!("Repair db failed, transaction_uid: {:?}, commit_uid: {:?}, reason: {:?}", transaciton_uid, commit_uid_copy, e))));
                                        return;
                                    }
                                } else {
                                    //无值，则执行删除操作
                                    if let Err(e) = tr.delete(vec![write]).await {
                                        //重播的删除操作失败，则立即返回错误原因
                                        let _ = sender.send(Err(Error::new(ErrorKind::Other, format!("Repair db failed, transaction_uid: {:?}, commit_uid: {:?}, reason: {:?}", transaciton_uid, commit_uid_copy, e))));
                                        return;
                                    }
                                }
                            }
                        }

                        //更新日志缓冲区偏移
                        offset = new_offset;
                    }

                    //指定本次重播事务的事务唯一id后执行预提交修复
                    if let Err(e) = tr.prepare_repair(transaciton_uid.clone()).await {
                        //预提交重播事务失败，则立即返回错误原因
                        let _ = sender.send(Err(Error::new(ErrorKind::Other,
                                                           format!("Repair db failed, transaction_uid: {:?}, commit_uid: {:?}, reason: {:?}",
                                                                   transaciton_uid,
                                                                   commit_uid_copy,
                                                                   e))));
                        return;
                    }

                    //指定本次重播事务的事务唯一id和事务提交唯一id后执行提交修复
                    if let Err(e) = tr
                        .commit_repair(transaciton_uid.clone(),
                                       commit_uid_copy.clone(),
                                       prepare_output).await {
                        //提交重播事务失败，则立即返回错误原因
                        let _ = sender.send(Err(Error::new(ErrorKind::Other,
                                                           format!("Repair db failed, transaction_uid: {:?}, commit_uid: {:?}, reason: {:?}",
                                                                   transaciton_uid,
                                                                   commit_uid_copy,
                                                                   e))));
                        return;
                    }

                    //返回成功重播一条未确认的提交日志
                    let _ = sender.send(Ok(()));
                } else {
                    //创建数据库事务失败，则立即返回错误原因
                    let _ = sender.send(Err(Error::new(ErrorKind::Other, format!("Repair db failed, transaction_uid: {:?}, commit_uid: {:?}, reason: get db transaction error", transaciton_uid, commit_uid_copy))));
                }
            }.boxed();
            let _ = db_mgr.0.rt.spawn(boxed);

            //注意单个线程的运行时重播或并发重播，则可能会发生阻塞
            match receiver.recv() {
                Err(e) => {
                    //同步通道异常，则立即返回错误原因
                    Err(Error::new(ErrorKind::Other, format!("Repair db failed, commit_uid: {:?}, reason: {:?}", commit_uid, e)))
                },
                Ok(result) => {
                    //同步阻塞的等待异步重播完成，则立即返回重播结果
                    result
                },
            }
        };

        //异步重播所有未确认的提交日志
        let replay_result = match self.0.tr_mgr.replay_commit_log(replay_callback).await {
            Err(e) if invalid_remove_data.load(Ordering::Acquire) => {
                // CommitLoggerLoader 当前丢失 callback kind，但原错误仍完整保存在其错误文本/源中。
                // 只对上面明确标记的删表持久化输入恢复 InvalidData，禁止按字符串猜测分类。
                return Err(Error::new(ErrorKind::InvalidData, e));
            },
            Err(e) => return Err(e),
            Ok(result) => result,
        };

        //所有未确认的提交日志已完成重播，则立即返回数据库修复成功
        let _ = self.0.tr_mgr.finish_replay().await?; //通知事务管理器，已完成重播

        // for table in tables.lock().await.keys() {
        //     if let Some(KVDBTable::BtreeOrdTab(tab)) = self.get_table(table).await {
        //         //当前表存在且为有序B树表，则立即整理
        //         tab.collect().await;
        //     }
        // }

        return Ok(replay_result);
    }
}

// 内部键值对数据库管理器
struct InnerKVDBManager<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    rt:                 MultiTaskRuntime<()>,                           //异步运行时
    tr_mgr:             Transaction2PcManager<C, Log>,                  //事务管理器
    db_path:            PathBuf,                                        //数据库的表文件所在目录的路径
    tables_meta_path:   PathBuf,                                        //数据库的元信息表文件所在目录的路径
    tables_path:        PathBuf,                                        //数据库表文件所在目录的路径
    tables:             Arc<RwLock<XHashMap<Atom, RegisteredTable<C, Log>>>>, //数据表及其精确版本状态
    key_versions:       KeyVersionRegistry,                             //数据库唯一的全局 Key 版本注册表
    status:             AtomicU64,                                      //数据库状态
    listener:           Option<Receiver<KVDBEvent<Guid>>>,              //数据库事件监听器
    notifier:           Option<Sender<KVDBEvent<Guid>>>,                //数据库事件通知器
}

#[derive(Clone)]
struct RegisteredTable<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    table: KVDBTable<C, Log>,
    versions: KeyVersions,
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> RegisteredTable<C, Log> {
    fn new(table: KVDBTable<C, Log>, versions: KeyVersions) -> Self {
        Self { table, versions }
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Deref for RegisteredTable<C, Log> {
    type Target = KVDBTable<C, Log>;

    fn deref(&self) -> &Self::Target {
        &self.table
    }
}

/// 在持有数据库表注册表写锁时安装表及其唯一版本状态。
///
/// 全局版本注册表先指向新实例，再替换表条目；旧事务仍可安全持有旧实例，但后续按身份删除
/// 旧实例时不会误删同名新表的版本状态。调用方必须持续持有 `tables` 的写锁，禁止把该顺序
/// 拆分到多个临界区。
fn install_registered_table<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
>(tables: &mut XHashMap<Atom, RegisteredTable<C, Log>>,
  key_versions: &KeyVersionRegistry,
  name: Atom,
  table: KVDBTable<C, Log>) {
    let versions = key_versions.create_table_versions();
    key_versions.install(name.clone(), versions.clone());
    if let Some(previous) = tables.insert(name.clone(), RegisteredTable::new(table, versions)) {
        key_versions.remove_exact(&name, &previous.versions);
    }
}

/// 根事务与五种表子事务共用的事务树节点枚举。
///
/// [`KVDBManager::transaction`] 只返回 [`Self::RootTr`]。应用层应始终通过该根 variant 调用
/// DDL、KV、stream、lock 和生命周期 API；其它 variant 由根事务在首次触表时惰性创建，供
/// `pi_async_transaction` 遍历事务树。许多公开 wrapper 在子表 variant 上会 panic，这些
/// variant 公开可构造并不代表它们属于合法应用调用域。
///
/// clone 为共享句柄 clone，不复制事务快照或状态。类型可在线程间移动/共享以支持 runtime
/// 调度，但同一逻辑事务上的并发动作、prepare/commit/rollback 越序调用不提供可串行化保证；
/// 当前状态防线不足见 `FIND-TR-002`。真实根生命周期见
/// `tests/root_transaction_lifecycle.rs`。
#[derive(Clone)]
pub enum KVDBTransaction<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    /// 应用层唯一合法的事务入口；持有数据库管理器、子事务树和根版本上下文。
    RootTr(RootTransaction<C, Log>),
    /// 根事务按需创建的 Meta 子事务节点。
    MetaTabTr(MetaTabTr<C, Log>),
    /// 根事务按需创建的 Memory 子事务节点。
    MemOrdTabTr(MemOrdTabTr<C, Log>),
    /// 根事务按需创建的 LogOrdered 子事务节点。
    LogOrdTabTr(LogOrdTabTr<C, Log>),
    /// 根事务按需创建的 LogWrite 子事务节点；当前不允许外部业务使用该表。
    LogWTabTr(LogWTabTr<C, Log>),
    /// 根事务按需创建的 Btree 子事务节点。
    BtreeOrdTabTr(BtreeOrdTabTr<C, Log>),
}

// SAFETY: 每个 variant 都是 Arc/同步原语保护的根或表事务共享句柄；移动枚举只移动该 owner，
// 不搬移自引用数据。各表事务实现负责其 COW root、cache、状态和存储句柄的跨线程同步，Log
// 又受 AsyncCommitLog: Send + Sync 约束。该保证只覆盖内存/线程安全，不扩大合法状态机域。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Send for KVDBTransaction<C, Log> {}
// SAFETY: 共享调用最终委托给 RootTransaction 或各表事务的同步字段；枚举自身没有额外内部
// 可变性、裸指针或 thread-owner 状态。并发逻辑动作仍须遵守事务状态和冲突协议。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Sync for KVDBTransaction<C, Log> {}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> AsyncTransaction for KVDBTransaction<C, Log> {
    type Output = ();
    type Error = KVTableTrError;

    fn is_writable(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_writable()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_writable()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_writable()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_writable()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_writable()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_writable()
            }
        }
    }

    // 键值对数据库的提交，会把所有子事务的预提交输出合成为一个提交输入，用于写入提交日志，所以也不需要并发
    fn is_concurrent_commit(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_concurrent_commit()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_concurrent_commit()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_concurrent_commit()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_concurrent_commit()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_concurrent_commit()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_concurrent_commit()
            },
        }
    }

    // 键值对数据库的预提交基本都是内存操作，所以回滚也不需要并发
    fn is_concurrent_rollback(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_concurrent_rollback()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_concurrent_rollback()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_concurrent_rollback()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_concurrent_rollback()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_concurrent_rollback()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_concurrent_rollback()
            },
        }
    }

    fn get_source(&self) -> Atom {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_source()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_source()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_source()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_source()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_source()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_source()
            },
        }
    }

    fn init(&self)
            -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.init()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.init()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.init()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.init()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.init()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.init()
            },
        }
    }

    fn rollback(&self)
                -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.rollback()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.rollback()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.rollback()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.rollback()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.rollback()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.rollback()
            },
        }
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Transaction2Pc for KVDBTransaction<C, Log> {
    type Tid = Guid;
    type Pid = Guid;
    type Cid = Guid;
    type PrepareOutput = Vec<u8>;
    type PrepareError = KVTableTrError;
    type ConfirmOutput = ();
    type ConfirmError = KVTableTrError;
    type CommitConfirm = KVDBCommitConfirm<C, Log>;

    fn is_require_persistence(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_require_persistence()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_require_persistence()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_require_persistence()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_require_persistence()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_require_persistence()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_require_persistence()
            },
        }
    }

    fn require_persistence(&self) {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.require_persistence();
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.require_persistence();
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.require_persistence();
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.require_persistence();
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.require_persistence();
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.require_persistence();
            },
        }
    }

    fn is_concurrent_prepare(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_concurrent_prepare()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_concurrent_prepare()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_concurrent_prepare()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_concurrent_prepare()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_concurrent_prepare()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_concurrent_prepare()
            },
        }
    }

    fn is_enable_inherit_uid(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_enable_inherit_uid()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_enable_inherit_uid()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_enable_inherit_uid()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_enable_inherit_uid()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_enable_inherit_uid()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_enable_inherit_uid()
            },
        }
    }

    fn get_transaction_uid(&self) -> Option<<Self as Transaction2Pc>::Tid> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_transaction_uid()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_transaction_uid()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_transaction_uid()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_transaction_uid()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_transaction_uid()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_transaction_uid()
            },
        }
    }

    fn set_transaction_uid(&self, uid: <Self as Transaction2Pc>::Tid) {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.set_transaction_uid(uid);
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.set_transaction_uid(uid);
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.set_transaction_uid(uid);
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.set_transaction_uid(uid);
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.set_transaction_uid(uid);
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.set_transaction_uid(uid);
            },
        }
    }

    fn get_prepare_uid(&self) -> Option<<Self as Transaction2Pc>::Pid> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_prepare_uid()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_prepare_uid()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_prepare_uid()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_prepare_uid()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_prepare_uid()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_prepare_uid()
            },
        }
    }

    fn set_prepare_uid(&self, uid: <Self as Transaction2Pc>::Pid) {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.set_prepare_uid(uid);
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.set_prepare_uid(uid);
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.set_prepare_uid(uid);
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.set_prepare_uid(uid);
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.set_prepare_uid(uid);
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.set_prepare_uid(uid);
            },
        }
    }

    fn get_commit_uid(&self) -> Option<<Self as Transaction2Pc>::Cid> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_commit_uid()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_commit_uid()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_commit_uid()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_commit_uid()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_commit_uid()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_commit_uid()
            },
        }
    }

    fn set_commit_uid(&self, uid: <Self as Transaction2Pc>::Cid) {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.set_commit_uid(uid);
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.set_commit_uid(uid);
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.set_commit_uid(uid);
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.set_commit_uid(uid);
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.set_commit_uid(uid);
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.set_commit_uid(uid);
            },
        }
    }

    fn get_prepare_timeout(&self) -> u64 {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_prepare_timeout()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_prepare_timeout()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_prepare_timeout()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_prepare_timeout()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_prepare_timeout()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_prepare_timeout()
            },
        }
    }

    fn get_commit_timeout(&self) -> u64 {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_commit_timeout()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_commit_timeout()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_commit_timeout()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_commit_timeout()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_commit_timeout()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_commit_timeout()
            },
        }
    }

    fn prepare(&self)
               -> BoxFuture<Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        #[cfg(feature = "default")]
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.prepare()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.prepare()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.prepare()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.prepare()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.prepare()
            },
        }
    }

    fn prepare_conflicts(&self)
        -> BoxFuture<Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        #[cfg(feature = "default")]
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare_conflicts()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.prepare_conflicts()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.prepare_conflicts()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.prepare_conflicts()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.prepare_conflicts()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.prepare_conflicts()
            },
        }
    }

    fn commit(&self, confirm: <Self as Transaction2Pc>::CommitConfirm)
              -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        #[cfg(feature = "default")]
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.commit(confirm)
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.commit(confirm)
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.commit(confirm)
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.commit(confirm)
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.commit(confirm)
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.commit(confirm)
            },
        }
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Transaction2PcAllConflicts for KVDBTransaction<C, Log> {
    fn precheck_all_conflicts(&self)
        -> BoxFuture<'_, Result<(), <Self as Transaction2Pc>::PrepareError>> {
        match self {
            KVDBTransaction::RootTr(tr) => tr.precheck_all_conflicts(),
            KVDBTransaction::MetaTabTr(tr) => tr.precheck_all_conflicts(),
            KVDBTransaction::MemOrdTabTr(tr) => tr.precheck_all_conflicts(),
            KVDBTransaction::LogOrdTabTr(tr) => tr.precheck_all_conflicts(),
            KVDBTransaction::LogWTabTr(tr) => tr.precheck_all_conflicts(),
            KVDBTransaction::BtreeOrdTabTr(tr) => tr.precheck_all_conflicts(),
        }
    }

    fn prepare_all_conflicts(&self)
        -> BoxFuture<'_, Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        match self {
            KVDBTransaction::RootTr(tr) => tr.prepare_all_conflicts(),
            KVDBTransaction::MetaTabTr(tr) => tr.prepare_all_conflicts(),
            KVDBTransaction::MemOrdTabTr(tr) => tr.prepare_all_conflicts(),
            KVDBTransaction::LogOrdTabTr(tr) => tr.prepare_all_conflicts(),
            KVDBTransaction::LogWTabTr(tr) => tr.prepare_all_conflicts(),
            KVDBTransaction::BtreeOrdTabTr(tr) => tr.prepare_all_conflicts(),
        }
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> UnitTransaction for KVDBTransaction<C, Log> {
    type Status = Transaction2PcStatus;
    type Qos = TableTrQos;

    fn is_unit(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_unit()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_unit()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_unit()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_unit()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_unit()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_unit()
            },
        }
    }

    fn get_status(&self) -> <Self as UnitTransaction>::Status {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.get_status()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.get_status()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.get_status()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.get_status()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.get_status()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.get_status()
            },
        }
    }

    fn set_status(&self, status: <Self as UnitTransaction>::Status) {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.set_status(status);
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.set_status(status);
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.set_status(status);
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.set_status(status);
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.set_status(status);
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.set_status(status);
            },
        }
    }

    fn qos(&self) -> <Self as UnitTransaction>::Qos {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.qos()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.qos()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.qos()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.qos()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.qos()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.qos()
            },
        }
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> SequenceTransaction for KVDBTransaction<C, Log> {
    type Item = Self;

    // 键值对数据表事务，一定不是顺序事务
    fn is_sequence(&self) -> bool {
        false
    }

    fn prev_item(&self) -> Option<<Self as SequenceTransaction>::Item> {
        None
    }

    fn next_item(&self) -> Option<<Self as SequenceTransaction>::Item> {
        None
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> TransactionTree for KVDBTransaction<C, Log> {
    type Node = KVDBTransaction<C, Log>; //键值对数据库的根事务的子事务，必须是键值对数据库事务
    type NodeInterator = KVDBChildTrList<C, Log>;

    fn is_tree(&self) -> bool {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.is_tree()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.is_tree()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.is_tree()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.is_tree()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.is_tree()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.is_tree()
            },
        }
    }

    fn children_len(&self) -> usize {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.children_len()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.children_len()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.children_len()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.children_len()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.children_len()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.children_len()
            },
        }
    }

    fn to_children(&self) -> Self::NodeInterator {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.to_children()
            },
            KVDBTransaction::MetaTabTr(tr) => {
                tr.to_children()
            },
            KVDBTransaction::MemOrdTabTr(tr) => {
                tr.to_children()
            },
            KVDBTransaction::LogOrdTabTr(tr) => {
                tr.to_children()
            },
            KVDBTransaction::LogWTabTr(tr) => {
                tr.to_children()
            },
            KVDBTransaction::BtreeOrdTabTr(tr) => {
                tr.to_children()
            },
        }
    }
}

/*
* 键值对数据库事务异步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> KVDBTransaction<C, Log> {
    /// 异步获取表的元信息
    pub async fn table_meta(&self, name: Atom) -> Option<KVTableMeta> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.table_meta(name).await
            },
            _ => panic!("Get table meta failed, reason: invalid root transaction"),
        }
    }

    /// 异步创建表，需要指定表名和表的元信息
    pub async fn create_table_with_options(&self,
                                           name: Atom,
                                           meta: KVTableMeta,
                                           options: CreateTableOptions,
                                           enable_accelerated_repair: bool) -> IOResult<()> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.create_table_with_options(
                    name,
                    meta,
                    options,
                    enable_accelerated_repair
                ).await
            },
            _ => panic!("Create table failed, reason: invalid root transaction"),
        }
    }

    /// 异步创建表，需要指定表名和表的元信息
    pub async fn create_table(&self,
                              name: Atom,
                              meta: KVTableMeta,
                              enable_accelerated_repair: bool) -> IOResult<()> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.create_table(
                    name,
                    meta,
                    enable_accelerated_repair
                ).await
            },
            _ => panic!("Create table failed, reason: invalid root transaction"),
        }
    }

    /// 异步批量创建表，只允许在初始化加载表时使用
    pub(crate) async fn create_multiple_tables(&self,
                                               table_metas: Vec<(Atom, KVTableMeta, Option<CreateTableOptions>)>,
                                               is_checksum: bool,
                                               enable_accelerated_repair: bool)
        -> IOResult<()>
    {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.create_multiple_tables(
                    table_metas,
                    is_checksum,
                    enable_accelerated_repair
                ).await
            },
            _ => panic!("Create multiple table failed, reason: invalid root transaction"),
        }
    }

    /// 异步修复创建表，需要指定表名和表的元信息
    pub(crate) async fn repair_create_table(&self,
                                            name: Atom,
                                            meta: KVTableMeta,
                                            enable_accelerated_repair: bool) -> IOResult<()> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.repair_create_table(name, meta, enable_accelerated_repair).await
            },
            _ => panic!("Create table failed, reason: invalid root transaction"),
        }
    }

    /// 从当前数据库根事务中移除指定表，并登记持久化的 Meta tombstone。
    ///
    /// `name` 的所有权移入本调用；其 UTF-8 编码长度必须位于
    /// `1..=`[`crate::MAX_TABLE_NAME_BYTES`]。空名或超长名称会在根持久化标记、表注册表和
    /// Meta 子事务发生任何变化前返回 [`std::io::ErrorKind::InvalidInput`]。内部 Meta 表名
    /// `.tables_meta` 同样会被无副作用拒绝，防止删除数据库自身的表目录。当前校验仍不拒绝
    /// 路径分隔符、绝对路径、`.` 或 `..`，调用方须遵守
    /// [CONTRACT-TABLE-NAME-001](../docs/SEMANTIC_CONTRACTS.md#contract-table-name-001) 中记录的
    /// 未决名称/路径边界。
    ///
    /// # 可观察顺序与返回值
    ///
    /// 对合法名称，当前实现按以下顺序执行：
    ///
    /// 1. 将根事务标记为需要持久化；
    /// 2. 获取数据库表注册表的异步写锁，并立即移除名称对应的注册项；
    /// 3. 在同一根事务的 Meta 子事务中写入该名称的删除动作；
    /// 4. 返回 `Ok(())`，由调用方随后执行 [`Self::prepare_modified`] 和
    ///    [`Self::commit_modified`]。
    ///
    /// `Ok(())` 只表示删表动作已经登记，不表示事务已经提交、根 WAL 已落地、Meta 数据文件
    /// 已持久化或表资源已释放。`prepare_modified` 会生成包含 Meta tombstone 的根 WAL 输入；
    /// `commit_modified` 成功才表示该 WAL 已 append/flush，之后 Meta 数据文件异步持久化，
    /// 最终成功信号再确认该 WAL。提交成功后立即崩溃时，启动恢复会从原 WAL 重放删除。
    ///
    /// 表不存在不是错误：调用仍返回 `Ok(())` 并登记 tombstone，因此不能根据返回值判断表
    /// 原先是否存在。重复调用在最终注册状态上是幂等的，但会重复产生事务动作/WAL 副作用，
    /// 不是物理副作用意义上的幂等操作。Meta 动作失败返回 `io::ErrorKind::Other`，此时注册表
    /// 可能已经改变。
    ///
    /// # 事务、取消与生命周期边界
    ///
    /// 本方法只支持 [`KVDBTransaction::RootTr`]；对任一表子事务 variant 调用会 panic。调用方
    /// 必须使用可写根事务；当前实现不在本入口拒绝只读事务，只读误用可能已经改变注册表却在
    /// prepare 时跳过持久化。当前 DDL 不具完整事务原子性：rollback 不会恢复已移除的注册项，
    /// 合法名称通过校验后取消 future 也可能留下根持久化标记或部分副作用。因此本方法不是
    /// rollback-safe 或 cancellation-safe，调用方不能把 `remove_table` 返回前后的中间状态当作
    /// 原子切换。
    ///
    /// 移除注册项不会删除表目录/数据文件，不会停止后台 collector，也不会强制释放已有
    /// `Arc`、表事务、redb/logfile 句柄或迭代流。已有流仍只在其创建事务存活期间按快照契约
    /// 使用；实际资源回收边界见 `FIND-LIFE-001`。这些限制属于当前实现事实，不是最终或最佳
    /// DDL 设计。
    ///
    /// # 并发、性能与安全
    ///
    /// 方法会等待 `async_lock::RwLock` 表注册表写锁，并在持有该 guard 时取得根事务的同步
    /// `childs_map` 锁以及 Meta 事务内部同步锁；当前动作阶段不执行文件 I/O、用户回调、FFI
    /// 或 V8 操作，但竞争会暂停同一注册表上的其它管理操作。并发 DDL 由这些锁和后续 Meta
    /// prepare 冲突检查约束，不提供跨事务串行化或完整原子性保证。本实现没有新增 `unsafe`，
    /// 不自行产生裸指针或跨运行时 owner；其 `Send/Sync` 边界继承根事务、表和 runtime 契约。
    ///
    /// 表注册表删除平均为 O(1)，Meta COW 删除为 O(log m)，名称编码为 O(name_len)；动作阶段
    /// 额外空间为 O(name_len + log m)。真正根 WAL append/flush 的时间和空间由后续 prepare/
    /// commit 的完整事务 payload 决定。与修复前错误跳过 WAL 的行为相比，成功删表提交会增加
    /// 必需的 WAL I/O；修复本身在动作热路径只增加一次 O(1) relaxed 原子存储。
    ///
    /// # 使用顺序
    ///
    /// 以下片段假定 `transaction` 是由 manager 创建的可写根事务：
    ///
    /// ```ignore
    /// transaction.remove_table(Atom::from("users")).await?;
    /// let prepare_output = transaction.prepare_modified().await?;
    /// transaction.commit_modified(prepare_output).await?;
    /// ```
    ///
    /// 正式语义和 BUG 证据见
    /// [CONTRACT-DDL-REMOVE-001](../docs/SEMANTIC_CONTRACTS.md#contract-ddl-remove-crash-durability)
    /// 与 [修复归档](../docs/DDL_REMOVE_DURABILITY_FIX.md#ddl-remove-fix-index)；真实四进程
    /// crash/replay/data-only 验证入口为 `tests/ddl_remove_crash_durability.rs`，名称边界入口为
    /// `tests/table_name_contract.rs`。
    pub async fn remove_table(&self, name: Atom) -> IOResult<()> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.remove_table(name).await
            },
            _ => panic!("Remove table failed, reason: invalid root transaction"),
        }
    }

    /// 在根 WAL 重放期间登记一个删表动作。
    ///
    /// `name` 必须来自已成功解码的 Meta tombstone；空名或超长持久化名称按损坏数据返回
    /// [`std::io::ErrorKind::InvalidData`]。有效输入复用正常删表流程，因此同样只允许根事务，
    /// 同样不删除物理表文件，也继承当前 DDL 非完整原子性。该入口只由 `try_repair` 内部使用，
    /// 不会重新 append 原 WAL；replay commit 负责发布恢复结果和最终确认原事务。
    pub(crate) async fn repair_remove_table(&self, name: Atom) -> IOResult<()> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.repair_remove_table(name).await
            },
            _ => panic!("Remove table failed, reason: invalid root transaction"),
        }
    }

    /// 在键值对数据库事务的根事务内，异步查询多个表和键的值的结果集，可能会查询到旧值
    pub async fn dirty_query(&self,
                             table_kv_list: Vec<TableKV>) -> Vec<Option<Binary>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.dirty_query(table_kv_list).await
            },
            _ => panic!("Query by dirty db failed, reason: invalid root transaction"),
        }
    }

    /// 在键值对数据库事务的根事务内，异步查询多个表和键的值的结果集
    pub async fn query(&self,
                       table_kv_list: Vec<TableKV>) -> Vec<Option<Binary>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.query(table_kv_list).await
            },
            _ => panic!("Query db failed, reason: invalid root transaction"),
        }
    }

    /// 在键值对数据库事务的根事务内，异步插入或更新指定多个表和键的值，插入或更新可能会被覆蓋
    pub async fn dirty_upsert(&self,
                              table_kv_list: Vec<TableKV>) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.dirty_upsert(table_kv_list).await
            },
            _ => panic!("Upsert db failed, reason: invalid root transaction"),
        }
    }

    /// 在键值对数据库事务的根事务内，异步插入或更新指定多个表和键的值
    pub async fn upsert(&self,
                        table_kv_list: Vec<TableKV>) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.upsert(table_kv_list).await
            },
            _ => panic!("Upsert db failed, reason: invalid root transaction"),
        }
    }

    /// 在根事务内按输入顺序脏删除多个表和 key，并返回各表实现当前能够取得的旧值。
    ///
    /// Btree 缓存值直接返回；已有 tombstone/重复删除返回 `None`；缓存完全缺席时同步读取
    /// 调用时 redb 快照。redb 读取错误会记录详细 error 日志并降级为 `None`，删除仍写
    /// tombstone 并可提交，因此 `None` 不能无条件解释为持久化存储中原本没有 key。
    /// redb 成功读取的值或逻辑不存在会进入独立 KeyState 冲突基线，但不会写入事务创建时的
    /// `cache_ref`；读取失败只能保留 `OverlayMissing`。prepare 先用 revision 捕获同值写/ABA，
    /// 再按 allocation 身份快路和 bytes 回退比较逻辑状态。每个缓存缺席 Key 的独立读取可能
    /// 短暂阻塞 worker。完整边界见 `CONTRACT-BTREE-DELETE-001`、
    /// `CONTRACT-BTREE-PREPARE-BASELINE-001`、`tests/btree_delete_old_value.rs` 和
    /// `tests/btree_redb_prepare_baseline.rs`。只能对根事务调用，否则 panic。
    pub async fn dirty_delete(&self,
                        table_kv_list: Vec<TableKV>)
                        -> Result<Vec<Option<Binary>>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.dirty_delete(table_kv_list).await
            },
            _ => panic!("Delete db failed, reason: invalid root transaction"),
        }
    }

    /// 在根事务内按输入顺序删除多个表和 key，并返回各表实现当前能够取得的旧值。
    ///
    /// Btree 缓存值直接返回；已有 tombstone/重复删除返回 `None`；缓存完全缺席时同步读取
    /// 调用时 redb 快照。redb 读取错误会记录详细 error 日志并降级为 `None`，删除仍写
    /// tombstone 并可提交，因此 `None` 不能无条件解释为持久化存储中原本没有 key。
    /// redb 成功读取的值或逻辑不存在会进入独立 KeyState 冲突基线，但不会写入事务创建时的
    /// `cache_ref`；读取失败只能保留 `OverlayMissing`。prepare 先用 revision 捕获同值写/ABA，
    /// 再按 allocation 身份快路和 bytes 回退比较逻辑状态。每个缓存缺席 Key 的独立读取可能
    /// 短暂阻塞 worker。完整边界见 `CONTRACT-BTREE-DELETE-001`、
    /// `CONTRACT-BTREE-PREPARE-BASELINE-001`、`tests/btree_delete_old_value.rs` 和
    /// `tests/btree_redb_prepare_baseline.rs`。只能对根事务调用，否则 panic。
    pub async fn delete(&self,
                        table_kv_list: Vec<TableKV>)
                        -> Result<Vec<Option<Binary>>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.delete(table_kv_list).await
            },
            _ => panic!("Delete db failed, reason: invalid root transaction"),
        }
    }

    /// 在根事务内创建指定表的关键字快照流。
    ///
    /// `table_name` 必须是当前已注册表；不存在时返回 `None`。`key` 是包含边界，`None`
    /// 表示从升序首项或降序末项开始；`descending=false/true` 分别选择升序/降序。只能对
    /// [`KVDBTransaction::RootTr`] 调用，在子表事务枚举上调用会 panic。
    ///
    /// 首次访问表会创建并注册一个非持久化子事务，因此本方法不是纯函数；重复调用复用该
    /// 子事务，但每次返回独立的创建时快照。调用返回后，本事务及其它事务可继续
    /// `upsert/delete`，旧流保持不变；流不提供可串行化、实时可见或 commit/rollback 绑定。
    /// 创建本流的根事务必须存活到流耗尽或被 drop，事务释放后继续 poll 属于非法用法。
    ///
    /// 返回项拥有 key；流可在线程间移动但应由单消费者 poll。提前 drop 会释放快照。
    /// Memory/Meta/LogOrdered 创建为 O(1) COW 克隆加 O(log n) 定位；Btree 还会在本方法
    /// 返回前同步建立 redb 读事务，可能短暂阻塞。底层流无错误 item，Btree 读错误当前可能
    /// 表现为空流或提前结束。完整边界见 `CONTRACT-ITER-001` 和
    /// `tests/iterator_snapshot_safety.rs`。
    pub async fn keys<'a>(&self,
                          table_name: Atom,
                          key: Option<Binary>,
                          descending: bool)
                          -> Option<BoxStream<'a, Binary>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.keys(table_name,
                        key,
                        descending).await
            },
            _ => panic!("Get db keys failed, table: {:?}, key: {:?}, descending: {:?}, reason: invalid root transaction", table_name.as_str(), key, descending),
        }
    }

    /// 在根事务内创建指定表的键值对快照流。
    ///
    /// 表查找、根事务前置条件、包含边界、方向、子事务注册副作用、创建事务生命周期、
    /// 并发修改、取消、错误和复杂度与 [`KVDBTransaction::keys`] 相同。每个 item 是创建时
    /// 快照中 owned `(key, value)`；随后 value 更新或删除不会改变既有流。
    pub async fn values<'a>(&self,
                            table_name: Atom,
                            key: Option<Binary>,
                            descending: bool) -> Option<BoxStream<'a, (Binary, Binary)>> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.values(table_name,
                          key,
                          descending).await
            },
            _ => panic!("Get db values failed, table: {:?}, key: {:?}, descending: {:?}, reason: invalid root transaction", table_name.as_str(), key, descending)
        }
    }

    /// 在键值对数据库事务的根事务内，锁住指定表的指定关键字
    pub async fn lock_key(&self,
                          table_name: Atom,
                          key: Binary) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.lock_key(table_name, key).await
            },
            _ => panic!("Lock table key failed, table: {:?}, key: {:?}, reason: invalid root transaction", table_name.as_str(), key),
        }
    }

    /// 在键值对数据库事务的根事务内，解锁指定表的指定关键字
    pub async fn unlock_key(&self,
                            table_name: Atom,
                            key: Binary) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.unlock_key(table_name, key).await
            },
            _ => panic!("Unlock table key failed, table: {:?}, key: {:?}, reason: invalid root transaction", table_name.as_str(), key),
        }
    }

    /// 初始化并预提交整棵根事务树，返回随后 commit 所需的 opaque 字节。
    ///
    /// 只能对 [`KVDBTransaction::RootTr`] 调用；对子表 variant 调用会 panic。首次调用先向
    /// `Transaction2PcManager` 注册根事务并分配事务 UID，再按首次触表顺序 prepare 子事务。
    /// 成功返回的 `Vec<u8>` 归调用方所有：需要根 WAL 时包含根事务 UID 和持久化子表动作；
    /// 只读、无持久化动作及部分已完成 DDL 快路可返回空 Vec。
    ///
    /// 返回字节必须被视为一次性 opaque token，未经修改原样传给同一事务的
    /// [`Self::commit_modified`]。当前实现尚未校验 token 与事务身份/最近 prepare 输出绑定，
    /// 见 `FIND-TR-003`；篡改、跨事务交换、截断、附加或重复使用都不属于合法调用域。
    ///
    /// 非 Fatal prepare 错误使事务树失败但可调用 [`Self::rollback_modified`]；Fatal 永不可
    /// rollback。timeout 字段当前不执行实际截止。调用会获取多个同步锁并可能执行表级异步
    /// 操作；取消 future 不是已冻结的自动 rollback，调用方不得假设 drop future 会注销事务。
    pub async fn prepare_modified(&self) -> Result<Vec<u8>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare_modified().await
            },
            _ => panic!("Prepare modified db failed, reason: invalid root transaction"),
        }
    }

    /// 以可定位冲突的路径初始化并预提交整棵根事务树。
    ///
    /// 调用顺序、RootTr 前置条件、输出所有权、timeout、取消和错误恢复边界与
    /// [`Self::prepare_modified`] 相同。差异是表实现检测到写冲突时返回
    /// [`KVTableTrError::Conflicts`]，其中保存首个冲突表名和 Key，错误等级为 Normal；多个
    /// 冲突不保证全部报告，也不保证跨表诊断顺序独立于首次触表顺序。
    ///
    /// 冲突发生在根 WAL append/flush 和子表根原子发布之前，因此在没有其它 Fatal 节点时
    /// 可以 rollback。真实 same-key 冲突和 manager 计数闭环由
    /// `tests/root_transaction_lifecycle.rs` 验证。
    pub async fn prepare_modified_conflicts(&self) -> Result<Vec<u8>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare_modified_conflicts().await
            },
            _ => panic!("Prepare conflicts db failed, reason: invalid root transaction"),
        }
    }

    /// 以外部实际读取的版本集合和最终写集合执行完整冲突预提交。
    ///
    /// 只能对全新、可写的 RootTr 调用，并且只能与 [`Self::commit_with_version`] 组成独立协议；
    /// 禁止与普通/dirty 动作、DDL、迭代器、普通 prepare/commit 混用。`read_set` 和 `write_set`
    /// 均可为空；各集合内部不允许重复 `(table, key)`，跨集合重叠合法且最终动作以 write 为准。
    /// `Some(value)` 是 upsert，`None` 是 delete；LogWrite 不支持 delete。
    ///
    /// 所有表名、Key、Value 长度及重复项会在 UID、子事务和共享状态副作用前检查。版本失配及
    /// 只读表身份失效返回确定性 [`KVTableTrError::AllConflicts`]；写表缺失/替换和参数错误返回
    /// 可 rollback 的 Normal Common。成功 token 是一次性 opaque 数据，只能原样传给同一事务。
    pub async fn prepare_with_version(&self,
                                      read_set: Vec<TableKeyVersion>,
                                      write_set: Vec<TableKV>)
        -> Result<Vec<u8>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare_with_version(read_set, write_set).await
            },
            _ => panic!("Prepare with version failed, reason: invalid root transaction"),
        }
    }

    /// 提交一次已经成功 prepare 的根事务。
    ///
    /// `prepare_output` 必须是同一事务最近一次成功 prepare 返回的完整 Vec，并且只能使用
    /// 一次。只能对 RootTr 调用；子表 variant 会 panic。未先 prepare、重复 commit、传入其它
    /// 事务或修改后的 token 都是非法调用；当前部分非法路径可能返回 Normal 错误，另一些路径
    /// 会在内部 UID `unwrap` 处 panic，不能依赖其防御表现。
    ///
    /// 对需要持久化且有有效输出的事务，本方法先 append 并 flush 根 WAL；只有 WAL 落地成功
    /// 后才发布各子表 COW 根并安排最终数据文件持久化。`Ok(())` 表示第一阶段“事务提交成功”，
    /// 不表示所有数据文件已完成，也不表示根 WAL 已确认或改名 `.bak`。第二阶段只有全部持久化
    /// 子表发出成功信号后才由 [`KVDBCommitConfirm`] 异步确认，见 `CONTRACT-TR-002`。
    ///
    /// 在尚未调用根 WAL append/flush 时产生的非 Fatal 逻辑失败可按状态 rollback；WAL 已成功
    /// 后的数据文件失败不能 rollback，WAL 保持未确认并由重启 repair 补齐。方法会执行异步文件
    /// I/O、同步锁和 runtime 任务投递，不保证取消安全；调用方必须等待明确结果并另行观察最终确认。
    ///
    /// 当前事务安全保证不覆盖根 WAL 自身因磁盘空间/配额、只读或故障文件系统、设备 I/O、
    /// runtime 拒绝任务或文件大小限制导致的 append/flush 失败。依赖层普通 `io::Error` 不保留
    /// 失败阶段和累计写入字节，因此这类错误下的 Normal、`LogCommitFailed` 或 rollback 成功
    /// 都不能证明 WAL 完全未落盘，也不能证明 checkpoint 与磁盘状态已经回到事务前。该限制与
    /// 空 prepare 输出无关，后者直接跳过 WAL I/O。完整现状链、外部处置边界和未来设计入口见
    /// `LIMIT-ROOT-WAL-IO-001`：`docs/ROOT_WAL_IO_FAILURE_BOUNDARY.md`。
    pub async fn commit_modified(&self,
                                 prepare_output: Vec<u8>) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.commit_modified(prepare_output).await
            },
            _ => panic!("Commit modified db failed, reason: invalid root transaction"),
        }
    }

    /// 提交已经由 [`Self::prepare_with_version`] 成功预提交的根事务，并返回本事务发布的版本。
    ///
    /// token 必须来自同一事务且只能使用一次。返回项只描述本事务自己的最终写入，不会在方法尾
    /// 重新读取可能已被后续事务推进的全局最新版本；顺序不属于稳定契约。Ok 表示根 WAL 已按需
    /// 落地且表数据/版本已经发布，不表示异步数据文件全部持久化或 WAL 已确认为 `.bak`。
    /// 任一提交错误只返回 Err 并丢弃部分回执；WAL 成功后的错误不可 rollback。
    /// 根 WAL 自身的环境/runtime I/O 失败不属于当前事务安全保证，不能从 Err 或回执为空推断
    /// WAL 未写入；其边界与 [`Self::commit_modified`] 完全相同。
    pub async fn commit_with_version(&self,
                                     prepare_output: Vec<u8>)
        -> Result<Vec<TableKeyVersion>, KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.commit_with_version(prepare_output).await
            },
            _ => panic!("Commit with version failed, reason: invalid root transaction"),
        }
    }

    /// 回滚处于可恢复失败状态的整棵事务树。
    ///
    /// 只能对 RootTr 调用；子表 variant 会 panic。本方法不是任意时刻可用的 cancel：当前
    /// `Transaction2PcManager` 只接受 ActionFailed、PrepareFailed 或 LogCommitFailed，且事务树
    /// 中不得存在 Fatal。Start、Prepared、Commited、CommitFailed 等其它状态调用会返回错误，
    /// 并可能把状态推进到 RollbackFailed。
    ///
    /// 成功 rollback 会丢弃未发布的子表 COW 修改并从 manager 注销根事务；因为合法回滚点在
    /// 根 WAL 成功落地和数据文件写入之前，不会撤销已提交数据。Fatal 永不可 rollback。方法
    /// 可能 await 子事务回滚并获取同步锁；成功后返回 `Ok(())`，失败保留错误等级和事务状态，
    /// 不应继续复用严重失败句柄。
    ///
    /// 上述安全结论只适用于当前受支持的事务失败域。若 `LogCommitFailed` 来源是根 WAL 的磁盘、
    /// 文件系统、设备、runtime 或文件大小限制错误，当前依赖链无法证明 0/部分/完整落盘状态，
    /// 即使本方法返回 `Ok(())` 也不承诺事务安全或 logger checkpoint 已清理；见
    /// `LIMIT-ROOT-WAL-IO-001`。
    pub async fn rollback_modified(&self) -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.rollback_modified().await
            },
            _ => panic!("Rollback modified db failed, reason: invalid root transaction"),
        }
    }

    /// 在键值对数据库事务的根事务内，异步预提交本次事务对键值对数据库的所有修复修改，不返回预提交的输出
    async fn prepare_repair(&self,
                            transaction_uid: Guid)
                           -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.prepare_repair(transaction_uid).await
            },
            _ => panic!("Repair prepare modified db failed, reason: invalid root transaction"),
        }
    }

    /// 键值对数据库事务的根事务内，异步提交本次事务对键值对数据库的所有修复修改
    async fn commit_repair(&self,
                           transaction_uid: Guid,
                           commit_uid: Guid,
                           prepare_output: Vec<u8>)
        -> Result<(), KVTableTrError> {
        match self {
            KVDBTransaction::RootTr(tr) => {
                tr.commit_repair(transaction_uid,
                                 commit_uid,
                                 prepare_output).await
            },
            _ => panic!("Repair commit modified db failed, reason: invalid root transaction"),
        }
    }
}

///
/// 键值对数据库的根事务的子事务列表
///
#[derive(Clone)]
pub struct KVDBChildTrList<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
>(VecDeque<KVDBTransaction<C, Log>>);

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Iterator for KVDBChildTrList<C, Log> {
    type Item = KVDBTransaction<C, Log>;

    fn next(&mut self) -> Option<Self::Item> {
        self.0.pop_front()
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> KVDBChildTrList<C, Log> {
    /// 构建一个键值对数据库的根事务的子事务列表
    #[inline]
    pub(crate) fn new() -> Self {
        KVDBChildTrList(VecDeque::default())
    }

    /// 获取子事务的数量
    #[inline]
    pub(crate) fn len(&self) -> usize {
        self.0.len()
    }

    /// 加入一个指定的子事务
    #[inline]
    pub(crate) fn join(&mut self, tr: KVDBTransaction<C, Log>) -> usize {
        self.0.push_back(tr);
        self.len()
    }
}

/// 一棵数据库事务树的共享根节点。
///
/// 根节点保存 source、事务/提交 UID、2PC 状态、可写/持久化标志、当前未执行的 timeout、按
/// 表名索引的子事务 map、按首次触表顺序排列的子事务 list，以及 manager clone。子事务只在
/// 首次访问对应表时创建，同表后续动作复用同一节点。
///
/// 应用通常不直接构造或匹配本类型，而通过 [`KVDBTransaction::RootTr`] 使用。clone 共享同一
/// 状态机和子事务，不创建独立事务。根持有 manager，prepare 后 manager registry 又持有根；
/// 正常 commit/rollback 的 `finish` 会解除 registry 边，非法/取消流程可能延长生命周期。
#[derive(Clone)]
pub struct RootTransaction<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
>(Arc<InnerRootTransaction<C, Log>>);

// SAFETY: InnerRootTransaction 的可变 UID/status/子事务容器由 SpinLock 保护，持久化标志为
// AtomicBool，其余字段构造后只读；manager 自身满足 Send。移动 Arc 不改变内部地址或别名。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Send for RootTransaction<C, Log> {}
// SAFETY: 共享引用只能通过上述锁/原子或线程安全 manager 访问可变状态。该 impl 保证内存
// 安全，不保证对同一根事务并发调用多个动作或生命周期方法具有事务级串行语义。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Sync for RootTransaction<C, Log> {}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> AsyncTransaction for RootTransaction<C, Log> {
    type Output = ();
    type Error = KVTableTrError;

    fn is_writable(&self) -> bool {
        self.0.writable
    }

    // 键值对数据库的提交，会把所有子事务的预提交输出合成为一个提交输入，用于写入提交日志，所以也不需要并发
    fn is_concurrent_commit(&self) -> bool {
        false
    }

    // 键值对数据库的预提交基本都是内存操作，所以回滚也不需要并发
    fn is_concurrent_rollback(&self) -> bool {
        false
    }

    fn get_source(&self) -> Atom {
        self.0.source.clone()
    }

    fn init(&self)
            -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        async move {
            Ok(())
        }.boxed()
    }

    fn rollback(&self)
                -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        async move {
            Ok(())
        }.boxed()
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Transaction2Pc for RootTransaction<C, Log> {
    type Tid = Guid;
    type Pid = Guid;
    type Cid = Guid;
    type PrepareOutput = Vec<u8>;
    type PrepareError = KVTableTrError;
    type ConfirmOutput = ();
    type ConfirmError = KVTableTrError;
    type CommitConfirm = KVDBCommitConfirm<C, Log>;

    // 键值对数据库的根事务
    fn is_require_persistence(&self) -> bool {
        self.0.persistence.load(Ordering::Relaxed)
    }

    fn require_persistence(&self) {
        self.0.persistence.store(true, Ordering::Relaxed);
    }

    // 键值对数据库的预提交基本都是内存操作，不需要并发
    fn is_concurrent_prepare(&self) -> bool {
        false
    }

    // 键值对数据库的根事务是根事务，要求所有子事务的事务相关唯一id与根事务相同
    fn is_enable_inherit_uid(&self) -> bool {
        true
    }

    fn get_transaction_uid(&self) -> Option<<Self as Transaction2Pc>::Tid> {
        self.0.tid.lock().clone()
    }

    fn set_transaction_uid(&self, uid: <Self as Transaction2Pc>::Tid) {
        *self.0.tid.lock() = Some(uid);
    }

    fn get_prepare_uid(&self) -> Option<<Self as Transaction2Pc>::Pid> {
        None
    }

    fn set_prepare_uid(&self, _uid: <Self as Transaction2Pc>::Pid) {}

    fn get_commit_uid(&self) -> Option<<Self as Transaction2Pc>::Cid> {
        self.0.cid.lock().clone()
    }

    fn set_commit_uid(&self, uid: <Self as Transaction2Pc>::Cid) {
        *self.0.cid.lock() = Some(uid);
    }

    fn get_prepare_timeout(&self) -> u64 {
        self.0.prepare_timeout
    }

    fn get_commit_timeout(&self) -> u64 {
        self.0.commit_timeout
    }

    // 预提交键值对数据库的根事务
    fn prepare(&self)
               -> BoxFuture<Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        async move {
            if self.is_require_persistence() {
                //本次键值对数据库的根事务，需要持久化，则写入本次键值对数据库的根事务的事务唯一id的预提交输出缓冲区
                let mut prepare_output_head = Vec::new();
                let transaction_uid: Guid = self.get_transaction_uid().unwrap();
                prepare_output_head.put_u128_le(transaction_uid.0); //写入事务唯一id

                Ok(Some(prepare_output_head))
            } else {
                //本次键值对数据库的根事务，不需要持久化，则立即返回
                Ok(None)
            }
        }.boxed()
    }

    fn prepare_conflicts(&self) -> BoxFuture<Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        self.prepare()
    }

    fn commit(&self, _confirm: <Self as Transaction2Pc>::CommitConfirm)
              -> BoxFuture<Result<<Self as AsyncTransaction>::Output, <Self as AsyncTransaction>::Error>> {
        // 根节点没有独立数据文件，最终持久化由各子表负责；因此根 commit 不调用确认器，
        // 也不贡献成功计数。事务树完成调度不等于根 WAL 已最终确认。详见 CONTRACT-CFM-001：
        // docs/SEMANTIC_CONTRACTS.md#contract-confirm-success-only。
        async move {
            Ok(())
        }.boxed()
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Transaction2PcAllConflicts for RootTransaction<C, Log> {
    fn precheck_all_conflicts(&self)
        -> BoxFuture<'_, Result<(), <Self as Transaction2Pc>::PrepareError>> {
        let tr = self.clone();
        async move {
            tr.check_version_table_identities().await
        }.boxed()
    }

    fn prepare_all_conflicts(&self)
        -> BoxFuture<'_, Result<Option<<Self as Transaction2Pc>::PrepareOutput>, <Self as Transaction2Pc>::PrepareError>> {
        let tr = self.clone();
        async move {
            tr.check_version_table_identities().await?;
            tr.prepare().await
        }.boxed()
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> UnitTransaction for RootTransaction<C, Log> {
    type Status = Transaction2PcStatus;
    type Qos = TableTrQos;

    //键值对数据库的根事务，一定不是单元事务
    fn is_unit(&self) -> bool {
        false
    }

    fn get_status(&self) -> <Self as UnitTransaction>::Status {
        self.0.status.lock().clone()
    }

    fn set_status(&self, status: <Self as UnitTransaction>::Status) {
        *self.0.status.lock() = status;
    }

    fn qos(&self) -> <Self as UnitTransaction>::Qos {
        TableTrQos::Safe
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> SequenceTransaction for RootTransaction<C, Log> {
    type Item = Self;

    // 键值对数据库的根事务，一定不是顺序事务
    fn is_sequence(&self) -> bool {
        false
    }

    fn prev_item(&self) -> Option<<Self as SequenceTransaction>::Item> {
        None
    }

    fn next_item(&self) -> Option<<Self as SequenceTransaction>::Item> {
        None
    }
}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> TransactionTree for RootTransaction<C, Log> {
    type Node = KVDBTransaction<C, Log>; //键值对数据库的根事务的子事务，必须是键值对数据库事务
    type NodeInterator = KVDBChildTrList<C, Log>;

    // 键值对数据库事务的根事务，一定是事务树
    fn is_tree(&self) -> bool {
        true
    }

    // 获取键值对数据库事务的子事务数量
    fn children_len(&self) -> usize {
        self.0.childs.lock().len()
    }

    // 获取键值对数据库事务的子事务迭代器
    fn to_children(&self) -> Self::NodeInterator {
        self.0.childs.lock().clone()
    }
}

/*
* 键值对数据库的根事务同步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> RootTransaction<C, Log> {
    // 获取根确认器期待的成功信号数：每个 persistence=true 子事务恰好计一次，根事务不计。
    // 只有读动作的可写事务会得到 0；此时确认器只作为 inert 参数传过事务树，任何节点都不得
    // 调用它，也不存在需要确认的根 WAL。该值的协议边界详见 CONTRACT-CFM-001：
    // docs/SEMANTIC_CONTRACTS.md#contract-confirm-success-only。
    fn persistent_children_len(&self) -> usize {
        let mut len = 0;
        for child in self.to_children() {
            if child.is_require_persistence() {
                len += 1;
            }
        }

        len
    }

    // 创建指定名称表的子事务
    // 注意表事务是否持久化，表示事务是否允许持久化，允许事务持久化表示这个事务的所有写操作会被写入提交日志
    fn table_transaction(&self,
                         name: Atom,
                         table: &RegisteredTable<C, Log>,
                         is_persistent: bool,
                         childes_map: &mut XHashMap<Atom, KVDBTransaction<C, Log>>)
                         -> KVDBTransaction<C, Log> {
        match &table.table {
            KVDBTable::MetaTab(tab) => {
                //创建元信息表的表事务，并作为子事务注册到根事务上
                let tr = MetaTabTr::new_managed(self.get_source(),
                                                self.is_writable(),
                                                is_persistent,
                                                self.get_prepare_timeout(),
                                                self.get_commit_timeout(),
                                                tab.clone(),
                                                table.versions.clone(),
                                                PrepareMode::Ordinary,
                                                XHashMap::default(),
                                                None,
                                                XHashMap::default());
                let table_tr = KVDBTransaction::MetaTabTr(tr);

                //注册到键值对数据库的根事务
                childes_map.insert(name, table_tr.clone());
                self.0.childs.lock().join(table_tr.clone());

                table_tr
            },
            KVDBTable::MemOrdTab(tab) => {
                //创建有序内存表的表事务，并作为子事务注册到根事务上
                let tr = MemOrdTabTr::new_managed(self.get_source(),
                                                  self.is_writable(),
                                                  is_persistent,
                                                  self.get_prepare_timeout(),
                                                  self.get_commit_timeout(),
                                                  tab.clone(),
                                                  table.versions.clone(),
                                                  PrepareMode::Ordinary,
                                                  XHashMap::default(),
                                                  None,
                                                  XHashMap::default());
                let table_tr = KVDBTransaction::MemOrdTabTr(tr);

                //注册到键值对数据库的根事务
                childes_map.insert(name, table_tr.clone());
                self.0.childs.lock().join(table_tr.clone());

                table_tr
            },
            KVDBTable::LogOrdTab(tab) => {
                let tr = LogOrdTabTr::new_managed(self.get_source(),
                                                  self.is_writable(),
                                                  is_persistent,
                                                  self.get_prepare_timeout(),
                                                  self.get_commit_timeout(),
                                                  tab.clone(),
                                                  table.versions.clone(),
                                                  PrepareMode::Ordinary,
                                                  XHashMap::default(),
                                                  None,
                                                  XHashMap::default());
                let table_tr = KVDBTransaction::LogOrdTabTr(tr);

                //注册到键值对数据库的根事务
                childes_map.insert(name, table_tr.clone());
                self.0.childs.lock().join(table_tr.clone());

                table_tr
            },
            KVDBTable::LogWTab(tab) => {
                let tr = LogWTabTr::new_managed(self.get_source(),
                                                self.is_writable(),
                                                is_persistent,
                                                self.get_prepare_timeout(),
                                                self.get_commit_timeout(),
                                                tab.clone(),
                                                table.versions.clone(),
                                                PrepareMode::Ordinary,
                                                XHashMap::default(),
                                                None,
                                                XHashMap::default());
                let table_tr = KVDBTransaction::LogWTabTr(tr);

                //注册到键值对数据库的根事务
                childes_map.insert(name, table_tr.clone());
                self.0.childs.lock().join(table_tr.clone());

                table_tr
            },
            KVDBTable::BtreeOrdTab(tab) => {
                let tr = BtreeOrdTabTr::new_managed(self.get_source(),
                                                    self.is_writable(),
                                                    is_persistent,
                                                    self.get_prepare_timeout(),
                                                    self.get_commit_timeout(),
                                                    tab.clone(),
                                                    table.versions.clone(),
                                                    PrepareMode::Ordinary,
                                                    XHashMap::default(),
                                                    None,
                                                    XHashMap::default());
                let table_tr = KVDBTransaction::BtreeOrdTabTr(tr);

                //注册到键值对数据库的根事务
                childes_map.insert(name, table_tr.clone());
                self.0.childs.lock().join(table_tr.clone());

                table_tr
            },
        }
    }

    /// 为独立版本协议一次性安装预期版本和最终动作，并按输入首次触表顺序注册子事务。
    fn versioned_table_transaction(&self,
                                   name: Atom,
                                   table: &RegisteredTable<C, Log>,
                                   is_persistent: bool,
                                   expected: XHashMap<Binary, Version>,
                                   receipt: VersionReceipt,
                                   actions: XHashMap<Binary, crate::KVActionLog>,
                                   childes_map: &mut XHashMap<Atom, KVDBTransaction<C, Log>>)
        -> KVDBTransaction<C, Log> {
        let table_tr = match &table.table {
            KVDBTable::MetaTab(tab) => {
                KVDBTransaction::MetaTabTr(MetaTabTr::new_managed(
                    self.get_source(),
                    self.is_writable(),
                    is_persistent,
                    self.get_prepare_timeout(),
                    self.get_commit_timeout(),
                    tab.clone(),
                    table.versions.clone(),
                    PrepareMode::Versioned,
                    expected,
                    Some(receipt),
                    actions))
            },
            KVDBTable::MemOrdTab(tab) => {
                KVDBTransaction::MemOrdTabTr(MemOrdTabTr::new_managed(
                    self.get_source(),
                    self.is_writable(),
                    is_persistent,
                    self.get_prepare_timeout(),
                    self.get_commit_timeout(),
                    tab.clone(),
                    table.versions.clone(),
                    PrepareMode::Versioned,
                    expected,
                    Some(receipt),
                    actions))
            },
            KVDBTable::LogOrdTab(tab) => {
                KVDBTransaction::LogOrdTabTr(LogOrdTabTr::new_managed(
                    self.get_source(),
                    self.is_writable(),
                    is_persistent,
                    self.get_prepare_timeout(),
                    self.get_commit_timeout(),
                    tab.clone(),
                    table.versions.clone(),
                    PrepareMode::Versioned,
                    expected,
                    Some(receipt),
                    actions))
            },
            KVDBTable::LogWTab(tab) => {
                KVDBTransaction::LogWTabTr(LogWTabTr::new_managed(
                    self.get_source(),
                    self.is_writable(),
                    is_persistent,
                    self.get_prepare_timeout(),
                    self.get_commit_timeout(),
                    tab.clone(),
                    table.versions.clone(),
                    PrepareMode::Versioned,
                    expected,
                    Some(receipt),
                    actions))
            },
            KVDBTable::BtreeOrdTab(tab) => {
                KVDBTransaction::BtreeOrdTabTr(BtreeOrdTabTr::new_managed(
                    self.get_source(),
                    self.is_writable(),
                    is_persistent,
                    self.get_prepare_timeout(),
                    self.get_commit_timeout(),
                    tab.clone(),
                    table.versions.clone(),
                    PrepareMode::Versioned,
                    expected,
                    Some(receipt),
                    actions))
            },
        };

        childes_map.insert(name, table_tr.clone());
        self.0.childs.lock().join(table_tr.clone());
        table_tr
    }
}

/*
* 键值对数据库的根事务异步方法
*/
impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> RootTransaction<C, Log> {
    /// 检查版本协议装配时固定的表身份；只读失效汇总为冲突，任何写表失效优先返回 Common。
    async fn check_version_table_identities(&self) -> Result<(), KVTableTrError> {
        let Some(context) = self.0.version_context.lock().clone() else {
            return Ok(());
        };
        let tables = self.0.db_mgr.0.tables.read().await;
        let mut conflicts = Vec::new();
        let mut invalid_write_table = None;

        for identity in &context.tables {
            let exact = match (&identity.versions, tables.get(&identity.name)) {
                (Some(expected), Some(current)) => expected.ptr_eq(&current.versions),
                _ => false,
            };
            if exact {
                continue;
            }
            if identity.has_write {
                if invalid_write_table.is_none() {
                    invalid_write_table = Some(identity.name.clone());
                }
            } else {
                for key in &identity.read_keys {
                    conflicts.push(TableKey {
                        table: identity.name.clone(),
                        key: key.clone(),
                    });
                }
            }
        }
        drop(tables);

        if let Some(table) = invalid_write_table {
            return Err(KVTableTrError::new_transaction_error(
                ErrorLevel::Normal,
                format!("Prepare with version failed, table: {:?}, reason: write table is missing or was replaced after transaction assembly",
                        table.as_str())));
        }
        if conflicts.is_empty() {
            Ok(())
        } else {
            Err(KVTableTrError::new_all_conflicts_error(conflicts))
        }
    }

    /// 异步获取表的元信息
    #[inline]
    async fn table_meta(&self, table: Atom) -> Option<KVTableMeta> {
        let meta_table = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        let result = self.query(vec![TableKV::new(meta_table.clone(),
                                                  table_to_binary(&table),
                                                  None)]).await;
        if let Some(binary) = &result[0] {
            //指定名称的表，已注册元信息
            Some(KVTableMeta::from(binary.clone()))
        } else {
            None
        }
    }

    /// 异步创建表，表名可以是用文件分隔符分隔的路径，但必须是相对路径，且不允许使用".."
    #[inline]
    async fn create_table_with_options(&self,
                                       name: Atom,
                                       meta: KVTableMeta,
                                       options: CreateTableOptions,
                                       enable_accelerated_repair: bool) -> IOResult<()>
    {
        // 必须先于根持久化标记、注册表写锁、Meta 修改及文件/目录创建拒绝非法名称。
        validate_table_name(&name, ErrorKind::InvalidInput, "create table")?;

        //检查待创建的指定名称的表是否存在
        let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        let mut tables = self.0.db_mgr.0.tables.write().await;

        self.require_persistence(); //创建表的操作，一定会创建元信息表事务，而元信息表事务是需要持久化的事务，则根事务也设置为需要持久化
        if tables.contains_key(&name) {
            //指定名称的表已存在
            if let Some(meta_table) = tables.get(&meta_table_name) {
                //元信息表存在，则获取元信息表事务，并查询指定表的元信息
                let mut childes_map = self.0.childs_map.lock();
                let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                    //元信息表的子事务存在
                    table_tr.clone()
                } else {
                    //元信息表的子事务不存在，则创建元信息表的事务，因为可能只是查询操作，所以初始化指定表的子事务为非持久化事务
                    self.table_transaction(meta_table_name.clone(), meta_table, false, &mut *childes_map)
                };

                if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                    if let Some(value) = tr.query(table_to_binary(&name)).await {
                        //指定名称的表的元信息存在
                        let table_meta = KVTableMeta::from(value);
                        if table_meta == meta {
                            //待创建表的名称与已存在的表相同，且元信息相同，则立即返回创建成功
                            return Ok(());
                        } else {
                            //待创建表的名称与已存在的表相同，但元信息不同
                            if table_meta.is_persistence() {
                                match tables.get(&name).map(|registered| &registered.table) {
                                    Some(KVDBTable::LogOrdTab(tab)) => {
                                        if tab.len() > 0 {
                                            //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                            return Err(Error::new(ErrorKind::AlreadyExists,
                                                                  format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                        }
                                    },
                                    Some(KVDBTable::LogWTab(tab)) => {
                                        if tab.len() > 0 {
                                            //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                            return Err(Error::new(ErrorKind::AlreadyExists,
                                                                  format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                        }
                                    },
                                    Some(KVDBTable::BtreeOrdTab(tab)) => {
                                        if tab.len() > 0 {
                                            //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                            return Err(Error::new(ErrorKind::AlreadyExists,
                                                                  format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                        }
                                    },
                                    _ => (),
                                }
                            }
                        }
                    } else {
                        //指定名称的表的元信息不存在，则立即返回错误原因
                        return Err(Error::new(ErrorKind::AlreadyExists,
                                              format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict and table meta not exist", name, meta)));
                    }
                } else {
                    //不是元信息表事务，则立即返回错误原因
                    return Err(Error::new(ErrorKind::AlreadyExists,
                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: invalid meta table transaction", name, meta)));
                }
            }
        }

        //待创建的指定名称的表不存在，则创建指定名称的表，并将表的元信息注册到元信息表
        match meta.table_type {
            KVDBTableType::MemOrdTab => {
                //创建一个有序内存表
                let table = MemoryOrderedTable::new(name.clone(),
                                                    meta.persistence);

                //注册创建的有序内存表
                install_registered_table(&mut *tables,
                                         &self.0.db_mgr.0.key_versions,
                                         name.clone(),
                                         KVDBTable::MemOrdTab(table));
            },
            KVDBTableType::LogOrdTab => {
                //创建一个有序日志表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                if let CreateTableOptions::LogOrdTab(log_file_limit, block_limit, load_buf_len) = options.clone() {
                    //有序日志表的选项
                    let table =
                        LogOrderedTable::new(self.0.db_mgr.0.rt.clone(),
                                             table_path,
                                             name.clone(),
                                             log_file_limit,
                                             block_limit,
                                             None,
                                             load_buf_len as u64,
                                             true,
                                             16 * 1024 * 1024,
                                             60 * 1000,
                                             self.0.db_mgr.0.notifier.clone()).await;

                    //注册创建的有序日志表
                    install_registered_table(&mut *tables,
                                             &self.0.db_mgr.0.key_versions,
                                             name.clone(),
                                             KVDBTable::LogOrdTab(table));
                } else {
                    //没有有序日志表的选项，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Create table failed, name: {:?}, meta: {:?}, options: {:?}, reason: invalid options",
                                                  name,
                                                  meta,
                                                  options)));
                }
            },
            KVDBTableType::LogWTab => {
                //创建一个只写日志表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                let table =
                    LogWriteTable::new(self.0.db_mgr.0.rt.clone(),
                                         table_path,
                                         name.clone(),
                                         512 * 1024 * 1024,
                                         2 * 1024 * 1024,
                                         None,
                                         2 * 1024 * 1024,
                                         true,
                                         16 * 1024 * 1024,
                                         60 * 1000).await;

                //注册创建的只写日志表
                install_registered_table(&mut *tables,
                                         &self.0.db_mgr.0.key_versions,
                                         name.clone(),
                                         KVDBTable::LogWTab(table));
            },
            KVDBTableType::BtreeOrdTab => {
                //创建一个有序B树表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                if let CreateTableOptions::BtreeOrdTab(cache_size, enable_compact) = options.clone() {
                    //有序日志表的选项
                    let table =
                        BtreeOrderedTable::new(self.0.db_mgr.0.rt.clone(),
                                               table_path,
                                               name.clone(),
                                               cache_size,
                                               enable_compact,
                                               1024 * 1024,
                                               60 * 1000,
                                               enable_accelerated_repair,
                                               self.0.db_mgr.0.notifier.clone()).await;

                    //注册创建的有序日志表
                    install_registered_table(&mut *tables,
                                             &self.0.db_mgr.0.key_versions,
                                             name.clone(),
                                             KVDBTable::BtreeOrdTab(table));
                } else {
                    //没有有序日志表的选项，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Create table failed, name: {:?}, meta: {:?}, options: {:?}, reason: invalid options",
                                                  name,
                                                  meta,
                                                  options)));
                }
            },
        }

        //注册表的元信息
        if let Some(meta_table) = tables.get(&meta_table_name) {
            let mut childes_map = self.0.childs_map.lock();
            let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                //元信息表的子事务存在，则设置子事务为需要持久化
                table_tr.require_persistence();
                table_tr.clone()
            } else {
                //元信息表的子事务不存在，则创建元信息表的事务，因为需要创建表，所以初始化元信息表的子事务为持久化事务
                self.table_transaction(meta_table_name, meta_table, true, &mut *childes_map)
            };

            if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                if let Err(e) = tr.upsert(table_to_binary(&name),
                                               Binary::from(meta.clone())).await {
                    //写入表的元信息失败，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: {:?}", name, meta, e)));
                }
            } else {
                //不是元信息表事务，则立即返回错误原因
                return Err(Error::new(ErrorKind::Other,
                                      format!("Create table failed, name: {:?}, meta: {:?}, reason: invalid meta table transaction", name, meta)));
            }
        }

        Ok(())
    }

    /// 通过默认参数异步创建表，表名可以是用文件分隔符分隔的路径，但必须是相对路径，且不允许使用".."
    #[inline]
    async fn create_table(&self,
                          name: Atom,
                          meta: KVTableMeta,
                          enable_accelerated_repair: bool) -> IOResult<()>
    {
        match meta.table_type {
            KVDBTableType::LogOrdTab => {
                self.create_table_with_options(name,
                                               meta,
                                               CreateTableOptions::LogOrdTab(512 * 1024 * 1024,
                                                                             2 * 1024 * 1024,
                                                                             2 * 1024 * 1024),
                                               enable_accelerated_repair)
                    .await
            },
            KVDBTableType::BtreeOrdTab => {
                self.create_table_with_options(name,
                                               meta,
                                               CreateTableOptions::BtreeOrdTab(16 * 1024 * 1024,
                                                                               true),
                                               enable_accelerated_repair)
                    .await
            },
            _ => {
                self.create_table_with_options(name,
                                               meta,
                                               CreateTableOptions::Empty,
                                               enable_accelerated_repair)
                    .await
            },
        }

    }

    /// 异步创建指定的多个表，表名可以是用文件分隔符分隔的路径，但必须是相对路径，且不允许使用".."
    async fn create_multiple_tables(&self,
                                    table_metas: Vec<(Atom, KVTableMeta, Option<CreateTableOptions>)>,
                                    is_checksum: bool,
                                    enable_accelerated_repair: bool)
        -> IOResult<()>
    {
        // 启动加载是该内部批量入口的唯一生产调用方。先校验完整输入，避免合法项已注册后
        // 才在后项发现损坏名称，造成部分加载副作用。
        for (name, _, _) in &table_metas {
            validate_table_name(name,
                                ErrorKind::InvalidData,
                                "load multiple table metadata")?;
        }

        //创建表的操作，一定会创建元信息表事务，而元信息表事务是需要持久化的事务，则根事务也设置为需要持久化
        self.require_persistence();

        //检查待创建的指定名称的表是否存在
        let mut require_create_tables = Vec::new();
        let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        {
            let tables = self
                .0
                .db_mgr
                .0
                .tables
                .read()
                .await;

            for (name, meta, options) in table_metas {
                if tables.contains_key(&name) {
                    //指定名称的表已存在
                    if let Some(meta_table) = tables.get(&meta_table_name) {
                        //元信息表存在，则获取元信息表事务，并查询指定表的元信息
                        let mut childes_map = self.0.childs_map.lock();
                        let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                            //元信息表的子事务存在
                            table_tr.clone()
                        } else {
                            //元信息表的子事务不存在，则创建元信息表的事务，因为可能只是查询操作，所以初始化指定表的子事务为非持久化事务
                            self.table_transaction(meta_table_name.clone(), meta_table, false, &mut *childes_map)
                        };

                        if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                            if let Some(value) = tr.query(table_to_binary(&name)).await {
                                //指定名称的表的元信息存在
                                let table_meta = KVTableMeta::from(value);
                                if table_meta == meta {
                                    //待创建表的名称与已存在的表相同，且元信息相同，则立即返回创建成功
                                    return Ok(());
                                } else {
                                    //待创建表的名称与已存在的表相同，但元信息不同
                                    if table_meta.is_persistence() {
                                        match tables.get(&name).map(|registered| &registered.table) {
                                            Some(KVDBTable::LogOrdTab(tab)) => {
                                                if tab.len() > 0 {
                                                    //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                                    return Err(Error::new(ErrorKind::AlreadyExists,
                                                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                                }
                                            },
                                            Some(KVDBTable::LogWTab(tab)) => {
                                                if tab.len() > 0 {
                                                    //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                                    return Err(Error::new(ErrorKind::AlreadyExists,
                                                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                                }
                                            },
                                            Some(KVDBTable::BtreeOrdTab(tab)) => {
                                                if tab.len() > 0 {
                                                    //已存在的同名表是持久化表，且元信息不同，且表中有记录，则表名冲突
                                                    return Err(Error::new(ErrorKind::AlreadyExists,
                                                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict", name, meta)));
                                                }
                                            },
                                            _ => (), //忽略内存表元信息的不同
                                        }
                                    }
                                }
                            } else {
                                //指定名称的表的元信息不存在，则立即返回错误原因
                                return Err(Error::new(ErrorKind::AlreadyExists,
                                                      format!("Create table failed, name: {:?}, meta: {:?}, reason: name conflict and table meta not exist", name, meta)));
                            }
                        } else {
                            //不是元信息表事务，则立即返回错误原因
                            return Err(Error::new(ErrorKind::AlreadyExists,
                                                  format!("Create table failed, name: {:?}, meta: {:?}, reason: invalid meta table transaction", name, meta)));
                        }
                    }
                }

                //记录需要创建的表
                require_create_tables.push((name, meta, options));
            }
        }

        //并发创建待创建的表
        let result = AsyncValue::new();
        let count = Arc::new(AtomicU64::new(require_create_tables.len() as u64));
        for (name, meta, options) in require_create_tables.clone() {
            let db_rt = self.0.db_mgr.0.rt.clone();
            let tables_path = self
                .0
                .db_mgr
                .0
                .tables_path
                .clone();
            let tables = self
                .0
                .db_mgr
                .0
                .tables
                .clone();
            let key_versions = self.0.db_mgr.0.key_versions.clone();
            let result_copy = result.clone();
            let count_copy = count.clone();

            let notifier = self.0.db_mgr.0.notifier.clone();
            let _ = self.0.db_mgr.0.rt.spawn(async move {
                //待创建的指定名称的表不存在，则创建指定名称的表，并将表的元信息注册到元信息表
                match meta.table_type {
                    KVDBTableType::MemOrdTab => {
                        //创建一个有序内存表
                        let table = MemoryOrderedTable::new(name.clone(),
                                                            meta.persistence);

                        //注册创建的有序内存表
                        let mut tables = tables.write().await;
                        install_registered_table(&mut *tables,
                                                 &key_versions,
                                                 name.clone(),
                                                 KVDBTable::MemOrdTab(table));
                    },
                    KVDBTableType::LogOrdTab => {
                        //创建一个有序日志表
                        let table_path = tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                        if let Some(CreateTableOptions::LogOrdTab(log_file_limit, block_limit, load_buf_len)) = options.clone() {
                            //有序日志表的选项
                            let table =
                                LogOrderedTable::new(db_rt,
                                                     table_path,
                                                     name.clone(),
                                                     log_file_limit,
                                                     block_limit,
                                                     None,
                                                     load_buf_len as u64,
                                                     is_checksum,
                                                     16 * 1024 * 1024,
                                                     60 * 1000,
                                                     notifier).await;

                            //注册创建的有序日志表
                            let mut tables = tables.write().await;
                            install_registered_table(&mut *tables,
                                                     &key_versions,
                                                     name.clone(),
                                                     KVDBTable::LogOrdTab(table));
                        } else {
                            //没有有序日志表的选项，则立即通知错误原因
                            result_copy.set(Err(Error::new(ErrorKind::Other,
                                                           format!("Create table failed, name: {:?}, meta: {:?}, options: {:?}, reason: invalid options",
                                                                   name,
                                                                   meta,
                                                                   options))));
                            return;
                        }
                    },
                    KVDBTableType::LogWTab => {
                        //创建一个只写日志表
                        let table_path = tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                        let table =
                            LogWriteTable::new(db_rt,
                                               table_path,
                                               name.clone(),
                                               512 * 1024 * 1024,
                                               2 * 1024 * 1024,
                                               None,
                                               2 * 1024 * 1024,
                                               is_checksum,
                                               16 * 1024 * 1024,
                                               60 * 1000).await;

                        //注册创建的只写日志表
                        let mut tables = tables.write().await;
                        install_registered_table(&mut *tables,
                                                 &key_versions,
                                                 name.clone(),
                                                 KVDBTable::LogWTab(table));
                    },
                    KVDBTableType::BtreeOrdTab => {
                        //创建一个有序B树表
                        let table_path = tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                        if let Some(CreateTableOptions::BtreeOrdTab(cache_size, enable_compact)) = options.clone() {
                            //有序日志表的选项
                            let table =
                                BtreeOrderedTable::new(db_rt,
                                                       table_path,
                                                       name.clone(),
                                                       cache_size,
                                                       enable_compact,
                                                       1024 * 1024,
                                                       60 * 1000,
                                                       enable_accelerated_repair,
                                                       notifier).await;

                            //注册创建的有序日志表
                            let mut tables = tables.write().await;
                            install_registered_table(&mut *tables,
                                                     &key_versions,
                                                     name.clone(),
                                                     KVDBTable::BtreeOrdTab(table));
                        } else {
                            //没有有序日志表的选项，则立即通知错误原因
                            result_copy.set(Err(Error::new(ErrorKind::Other,
                                                           format!("Create table failed, name: {:?}, meta: {:?}, options: {:?}, reason: invalid options",
                                                                   name,
                                                                   meta,
                                                                   options))));
                            return;
                        }
                    },
                }

                if count_copy.fetch_sub(1, Ordering::AcqRel) <= 1 {
                    //本次所有待创建的表都已创建成功，则立即通知创建成功
                    result_copy.set(Ok(()));
                }
            });
        }

        //等待批量创建表全部成功
        result.await?;

        //注册表的元信息
        let mut tables = self
            .0
            .db_mgr
            .0
            .tables
            .write()
            .await;
        for (name, meta, _options) in require_create_tables {
            if let Some(meta_table) = tables.get(&meta_table_name) {
                let mut childes_map = self.0.childs_map.lock();
                let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                    //元信息表的子事务存在，则设置子事务为需要持久化
                    table_tr.require_persistence();
                    table_tr.clone()
                } else {
                    //元信息表的子事务不存在，则创建元信息表的事务，因为需要创建表，所以初始化元信息表的子事务为持久化事务
                    self.table_transaction(meta_table_name.clone(),
                                           meta_table,
                                           true,
                                           &mut *childes_map)
                };

                if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                    if let Err(e) = tr.upsert(table_to_binary(&name),
                                              Binary::from(meta.clone())).await {
                        //写入表的元信息失败，则立即返回错误原因
                        return Err(Error::new(ErrorKind::Other,
                                              format!("Create table failed, name: {:?}, meta: {:?}, reason: {:?}", name, meta, e)));
                    }
                } else {
                    //不是元信息表事务，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: invalid meta table transaction", name, meta)));
                }
            }
        }

        Ok(())
    }

    /// 异步修复创建表，表名可以是用文件分隔符分隔的路径，但必须是相对路径，且不允许使用".."
    #[inline]
    async fn repair_create_table(&self,
                                 name: Atom,
                                 meta: KVTableMeta,
                                 enable_accelerated_repair: bool) -> IOResult<()>
    {
        // 名称来自已落地根 WAL；非法长度表示持久化数据损坏，而不是本次调用参数错误。
        validate_table_name(&name, ErrorKind::InvalidData, "repair table creation")?;

        //检查待创建的指定名称的表是否存在
        let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        let mut tables = self.0.db_mgr.0.tables.write().await;

        self.0.persistence.store(true, Ordering::Relaxed); //创建表的操作，一定会创建元信息表事务，而元信息表事务是需要持久化的事务，则根事务也设置为需要持久化

        //待创建的指定名称的表不存在，则创建指定名称的表，并将表的元信息注册到元信息表
        match meta.table_type {
            KVDBTableType::MemOrdTab => {
                //创建一个有序内存表
                let table = MemoryOrderedTable::new(name.clone(),
                                                    meta.persistence);

                //注册创建的有序内存表
                install_registered_table(&mut *tables,
                                         &self.0.db_mgr.0.key_versions,
                                         name.clone(),
                                         KVDBTable::MemOrdTab(table));
            },
            KVDBTableType::LogOrdTab => {
                //创建一个有序日志表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                let table =
                    LogOrderedTable::new(self.0.db_mgr.0.rt.clone(),
                                         table_path,
                                         name.clone(),
                                         512 * 1024 * 1024,
                                         2 * 1024 * 1024,
                                         None,
                                         2 * 1024 * 1024,
                                         true,
                                         16 * 1024 * 1024,
                                         60 * 1000,
                                         self.0.db_mgr.0.notifier.clone()).await;

                //注册创建的有序日志表
                install_registered_table(&mut *tables,
                                         &self.0.db_mgr.0.key_versions,
                                         name.clone(),
                                         KVDBTable::LogOrdTab(table));
            },
            KVDBTableType::LogWTab => {
                //创建一个只写日志表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                let table =
                    LogWriteTable::new(self.0.db_mgr.0.rt.clone(),
                                       table_path,
                                       name.clone(),
                                       512 * 1024 * 1024,
                                       2 * 1024 * 1024,
                                       None,
                                       2 * 1024 * 1024,
                                       true,
                                       16 * 1024 * 1024,
                                       60 * 1000).await;

                //注册创建的只写日志表
                install_registered_table(&mut *tables,
                                         &self.0.db_mgr.0.key_versions,
                                         name.clone(),
                                         KVDBTable::LogWTab(table));
            },
            KVDBTableType::BtreeOrdTab => {
                //尝试创建一个有序B树表
                let table_path = self.0.db_mgr.0.tables_path.join(name.as_str()); //通过键值对数据库的表所在目录的路径与表名，生成表所在目录的路径
                if let Some(table) =
                    BtreeOrderedTable::try_new(self.0.db_mgr.0.rt.clone(),
                                               table_path,
                                               name.clone(),
                                               DEFAULT_CACHE_SIZE,
                                               true,
                                               1024 * 1024,
                                               60 * 1000,
                                               enable_accelerated_repair,
                                               self.0.db_mgr.0.notifier.clone()).await {
                    //尝试创建成功，则注册创建的有序日志表
                    install_registered_table(&mut *tables,
                                             &self.0.db_mgr.0.key_versions,
                                             name.clone(),
                                             KVDBTable::BtreeOrdTab(table));
                }
            },
        }

        //注册表的元信息
        if let Some(meta_table) = tables.get(&meta_table_name) {
            let mut childes_map = self.0.childs_map.lock();
            let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                //元信息表的子事务存在，则设置子事务为需要持久化
                table_tr.require_persistence();
                table_tr.clone()
            } else {
                //元信息表的子事务不存在，则创建元信息表的事务，因为需要创建表，所以初始化元信息表的子事务为持久化事务
                self.table_transaction(meta_table_name, meta_table, true, &mut *childes_map)
            };

            if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                if let Err(e) = tr.upsert(table_to_binary(&name),
                                          Binary::from(meta.clone())).await {
                    //写入表的元信息失败，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Create table failed, name: {:?}, meta: {:?}, reason: {:?}", name, meta, e)));
                }
            } else {
                //不是元信息表事务，则立即返回错误原因
                return Err(Error::new(ErrorKind::Other,
                                      format!("Create table failed, name: {:?}, meta: {:?}, reason: invalid meta table transaction", name, meta)));
            }
        }

        Ok(())
    }

    /// 实现根事务的删表动作；公开契约见 [`KVDBTransaction::remove_table`]。
    ///
    /// 根持久化标记必须先于任何注册表副作用，Meta tombstone 必须继续使用
    /// `table_to_binary` 的 BON Atom 编码。不得把本方法扩展为物理目录删除或 DDL rollback
    /// 补偿；两者均超出 BUG-DDL-REMOVE-001 的冻结修复边界。
    #[inline]
    async fn remove_table(&self, table: Atom) -> IOResult<()> {
        // 必须先于注册表移除和 Meta tombstone 拒绝非法名称，保证 InvalidInput 无副作用。
        validate_removable_table_name(&table, ErrorKind::InvalidInput, "remove table")?;

        // 删表会持久化 Meta tombstone，因此根事务必须参与 WAL。该标记必须先于注册表移除；
        // 否则 Meta 子事务虽会异步写数据文件，根 prepare 却会丢弃子日志，commit 返回成功后
        // 立即崩溃将无法恢复删除。这里只恢复删表耐久性，不改变 DDL 当前 rollback 非原子性。
        self.require_persistence();

        let mut tables = self.0.db_mgr.0.tables.write().await;

        //移除表
        if let Some(removed) = tables.remove(&table) {
            self.0
                .db_mgr
                .0
                .key_versions
                .remove_exact(&table, &removed.versions);
        }

        //删除表的元信息
        let meta_table_name = Atom::from(DEFAULT_DB_TABLES_META_DIR);
        if let Some(meta_table) = tables.get(&meta_table_name) {
            //元信息表存在，则获取元信息表事务，并查询指定表的元信息
            let mut childes_map = self.0.childs_map.lock();
            let meta_table_tr = if let Some(table_tr) = childes_map.get(&meta_table_name) {
                //元信息表的子事务存在，则设置子事务为需要持久化
                table_tr.require_persistence();
                table_tr.clone()
            } else {
                //元信息表的子事务不存在，则创建元信息表的事务，因为需要移除表，所以初始化元信息表的子事务为持久化事务
                self.table_transaction(meta_table_name, meta_table, true, &mut *childes_map)
            };

            if let KVDBTransaction::MetaTabTr(tr) = &meta_table_tr {
                if let Err(e) = tr.delete(table_to_binary(&table)).await {
                    //删除表的元信息失败，则立即返回错误原因
                    return Err(Error::new(ErrorKind::Other,
                                          format!("Remove table failed, name: {:?}, reason: {:?}", table, e)));
                }
            } else {
                //不是元信息表事务，则立即返回错误原因
                return Err(Error::new(ErrorKind::Other,
                                      format!("Remove table failed, name: {:?}, reason: invalid meta table transaction", table)));
            }
        }

        Ok(())
    }

    /// 重放已提交但未最终确认的删表 WAL；只接受有效持久化名称并复用正常删除动作。
    #[inline]
    async fn repair_remove_table(&self, table: Atom) -> IOResult<()> {
        // 恢复入口接收持久化 WAL 中的名称；先用持久化数据错误语义校验，再复用正常移除。
        validate_removable_table_name(&table, ErrorKind::InvalidData, "repair table removal")?;
        self.remove_table(table).await
    }

    /// 异步查询多个表和键的值的结果集，可能会查询到旧值
    #[inline]
    async fn dirty_query(&self,
                         table_kv_list: Vec<TableKV>) -> Vec<Option<Binary>> {
        let mut result = Vec::new();

        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始查询表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务，因为是查询操作，所以初始化指定表的子事务为非持久化事务
                    self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                };

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //查询元信息表的指定关键字的值
                        let value = tr.dirty_query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //查询有序内存表的指定关键字的值
                        let value = tr.dirty_query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //查询有序日志表的指定关键字的值
                        let value = tr.dirty_query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //查询只写日志表的指定关键字的值
                        let value = tr.dirty_query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //查询有序B树表的指定关键字的值
                        let value = tr.dirty_query(table_kv.key).await;
                        result.push(value);
                    },
                }
            } else {
                //指定名称的表不存在
                result.push(None);
            }
        }

        result
    }

    /// 异步查询多个表和键的值的结果集
    #[inline]
    async fn query(&self,
                   table_kv_list: Vec<TableKV>) -> Vec<Option<Binary>> {
        let mut result = Vec::new();

        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始查询表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务，因为是查询操作，所以初始化指定表的子事务为非持久化事务
                    self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                };

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //查询元信息表的指定关键字的值
                        let value = tr.query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //查询有序内存表的指定关键字的值
                        let value = tr.query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //查询有序日志表的指定关键字的值
                        let value = tr.query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //查询只写日志表的指定关键字的值
                        let value = tr.query(table_kv.key).await;
                        result.push(value);
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //查询有序B树表的指定关键字的值
                        let value = tr.query(table_kv.key).await;
                        result.push(value);
                    },
                }
            } else {
                //指定名称的表不存在
                result.push(None);
            }
        }

        result
    }

    /// 异步插入或更新指定多个表和键的值，插入或更新可能会被覆蓋
    #[inline]
    async fn dirty_upsert(&self,
                          table_kv_list: Vec<TableKV>) -> Result<(), KVTableTrError> {
        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始插入或更新表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以设置子事务为需要持久化
                        table_tr.require_persistence();
                    }
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以初始化指定表的子事务为持久化事务
                        self.table_transaction(table_kv.table, table, true, &mut *childes_map)
                    } else {
                        //指定表不需要持久化，所以即使插入或更新操作，也初始化指定表的子事务为非持久化事务
                        self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                    }
                };

                if table_tr.is_require_persistence() {
                    //如果任意写操作对应的子事务需要持久化，则根事务也需要持久化
                    self.0.persistence.store(true, Ordering::Relaxed);
                }

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //插入或更新元信息表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.dirty_upsert(table_kv.key, value).await {
                                //插入或更新元信息表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //插入或更新有序内存表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.dirty_upsert(table_kv.key, value).await {
                                //插入或更新有序内存表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //插入或更新有序日志表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.dirty_upsert(table_kv.key, value).await {
                                //插入或更新有序日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //插入或更新只写日志表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.dirty_upsert(table_kv.key, value).await {
                                //插入或更新只写日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //插入或更新有序B树表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.dirty_upsert(table_kv.key, value).await {
                                //插入或更新有序B树表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                }
            }
        }

        Ok(())
    }

    /// 异步插入或更新指定多个表和键的值
    #[inline]
    async fn upsert(&self,
                    table_kv_list: Vec<TableKV>) -> Result<(), KVTableTrError> {
        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始插入或更新表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以设置子事务为需要持久化
                        table_tr.require_persistence();
                    }
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以初始化指定表的子事务为持久化事务
                        self.table_transaction(table_kv.table, table, true, &mut *childes_map)
                    } else {
                        //指定表不需要持久化，所以即使插入或更新操作，也初始化指定表的子事务为非持久化事务
                        self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                    }
                };

                if table_tr.is_require_persistence() {
                    //如果任意写操作对应的子事务需要持久化，则根事务也需要持久化
                    self.0.persistence.store(true, Ordering::Relaxed);
                }

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //插入或更新元信息表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.upsert(table_kv.key, value).await {
                                //插入或更新元信息表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //插入或更新有序内存表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.upsert(table_kv.key, value).await {
                                //插入或更新有序内存表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //插入或更新有序日志表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.upsert(table_kv.key, value).await {
                                //插入或更新有序日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //插入或更新只写日志表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.upsert(table_kv.key, value).await {
                                //插入或更新只写日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //插入或更新有序B树表的指定关键字的值
                        if let Some(value) = table_kv.value {
                            //有值则插入或更新
                            if let Err(e) = tr.upsert(table_kv.key, value).await {
                                //插入或更新有序B树表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            }
                        }
                    },
                }
            } else {
                //指定名称的表不存在
                error!("Upsert table failed, table: {:?}, reason: table not exist",
                    &table_kv.table.as_str());
                return Err(KVTableTrError::new_transaction_error(ErrorLevel::Fatal,
                                                                 format!("Upsert table failed, table: {:?}, reason: table not exist",
                                                                     &table_kv.table.as_str())));
            }
        }

        Ok(())
    }

    /// 异步删除指定多个表和键的值，并返回删除值的结果集，删除可能会被覆蓋
    #[inline]
    async fn dirty_delete(&self,
                          table_kv_list: Vec<TableKV>) -> Result<Vec<Option<Binary>>, KVTableTrError> {
        let mut result = Vec::new();

        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始删除表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以设置子事务为需要持久化
                        table_tr.require_persistence();
                    }
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以初始化指定表的子事务为持久化事务
                        self.table_transaction(table_kv.table, table, true, &mut *childes_map)
                    } else {
                        //指定表不需要持久化，所以即使插入或更新操作，也初始化指定表的子事务为非持久化事务
                        self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                    }
                };

                if table_tr.is_require_persistence() {
                    //如果任意写操作对应的子事务需要持久化，则根事务也需要持久化
                    self.0.persistence.store(true, Ordering::Relaxed);
                }

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //删除元信息表的指定关键字的值
                        match tr.dirty_delete(table_kv.key).await {
                            Err(e) => {
                                //删除元信息表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除元信息表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //删除有序内存表的指定关键字的值
                        match tr.dirty_delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序内存表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序内存表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //删除有序日志表的指定关键字的值
                        match tr.dirty_delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序日志表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //删除只写日志表的指定关键字的值
                        match tr.dirty_delete(table_kv.key).await {
                            Err(e) => {
                                //删除只写日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除只写日志表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //删除有序B树表的指定关键字的值
                        match tr.dirty_delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序B树表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序B树表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                }
            } else {
                //指定名称的表不存在
                result.push(None);
            }
        }

        Ok(result)
    }

    /// 异步删除指定多个表和键的值，并返回删除值的结果集
    #[inline]
    async fn delete(&self,
                    table_kv_list: Vec<TableKV>) -> Result<Vec<Option<Binary>>, KVTableTrError> {
        let mut result = Vec::new();

        for table_kv in table_kv_list {
            if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_kv.table) {
                //指定名称的表存在，则获取表事务，并开始删除表的指定关键字的值
                let mut childes_map = self.0.childs_map.lock();
                let table_tr = if let Some(table_tr) = childes_map.get(&table_kv.table) {
                    //指定名称的表的子事务存在
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以设置子事务为需要持久化
                        table_tr.require_persistence();
                    }
                    table_tr.clone()
                } else {
                    //指定名称的表的子事务不存在，则创建指定表的事务
                    if table.is_persistent() {
                        //指定表需要持久化，且因为插入或更新操作，所以初始化指定表的子事务为持久化事务
                        self.table_transaction(table_kv.table, table, true, &mut *childes_map)
                    } else {
                        //指定表不需要持久化，所以即使插入或更新操作，也初始化指定表的子事务为非持久化事务
                        self.table_transaction(table_kv.table, table, false, &mut *childes_map)
                    }
                };

                if table_tr.is_require_persistence() {
                    //如果任意写操作对应的子事务需要持久化，则根事务也需要持久化
                    self.0.persistence.store(true, Ordering::Relaxed);
                }

                match &table_tr {
                    KVDBTransaction::RootTr(_tr) => {
                        //忽略键值对数据库的根事务
                        ()
                    },
                    KVDBTransaction::MetaTabTr(tr) => {
                        //删除元信息表的指定关键字的值
                        match tr.delete(table_kv.key).await {
                            Err(e) => {
                                //删除元信息表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除元信息表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::MemOrdTabTr(tr) => {
                        //删除有序内存表的指定关键字的值
                        match tr.delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序内存表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序内存表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::LogOrdTabTr(tr) => {
                        //删除有序日志表的指定关键字的值
                        match tr.delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序日志表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::LogWTabTr(tr) => {
                        //删除只写日志表的指定关键字的值
                        match tr.delete(table_kv.key).await {
                            Err(e) => {
                                //删除只写日志表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除只写日志表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                    KVDBTransaction::BtreeOrdTabTr(tr) => {
                        //删除有序B树表的指定关键字的值
                        match tr.delete(table_kv.key).await {
                            Err(e) => {
                                //删除有序B树表的指定关键字的值错误，则立即返回错误原因
                                return Err(e);
                            },
                            Ok(value) => {
                                //删除有序B树表的指定关键字的值成功
                                result.push(value);
                            },
                        }
                    },
                }
            } else {
                //指定名称的表不存在
                result.push(None);
            }
        }

        Ok(result)
    }

    /// 获取从指定表和关键字开始，从前向后或从后向前的关键字异步流
    #[inline]
    async fn keys<'a>(&self,
                      table_name: Atom,
                      key: Option<Binary>,
                      descending: bool) -> Option<BoxStream<'a, Binary>> {
        if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_name) {
            //指定名称的表存在，则获取表事务，并开始获取关键字的异步流
            let mut childes_map = self.0.childs_map.lock();
            let table_tr = if let Some(table_tr) = childes_map.get(&table_name) {
                //指定名称的表的子事务存在
                table_tr.clone()
            } else {
                //指定名称的表的子事务不存在，则创建指定表的事务，因为是查询操作，所以初始化指定表的子事务为非持久化事务
                self.table_transaction(table_name, table, false, &mut *childes_map)
            };

            match &table_tr {
                KVDBTransaction::RootTr(_tr) => {
                    //忽略键值对数据库的根事务
                    None
                },
                KVDBTransaction::MetaTabTr(tr) => {
                    //获取元信息表的关键字的异步流
                    Some(tr.keys(key, descending))
                },
                KVDBTransaction::MemOrdTabTr(tr) => {
                    //获取有序内存表的关键字的异步流
                    Some(tr.keys(key, descending))
                },
                KVDBTransaction::LogOrdTabTr(tr) => {
                    //获取有序日志表的关键字的异步流
                    Some(tr.keys(key, descending))
                },
                KVDBTransaction::LogWTabTr(tr) => {
                    //获取只写日志表的关键字的异步流
                    Some(tr.keys(key, descending))
                },
                KVDBTransaction::BtreeOrdTabTr(tr) => {
                    //获取有序B树表的关键字的异步流
                    Some(tr.keys(key, descending))
                },
            }
        } else {
            //指定名称的表不存在
            None
        }
    }

    /// 获取从指定表和关键字开始，从前向后或从后向前的键值对异步流
    #[inline]
    async fn values<'a>(&self,
                        table_name: Atom,
                        key: Option<Binary>,
                        descending: bool) -> Option<BoxStream<'a, (Binary, Binary)>> {
        if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_name) {
            //指定名称的表存在，则获取表事务，并开始获取键值对异步流
            let mut childes_map = self.0.childs_map.lock();
            let table_tr = if let Some(table_tr) = childes_map.get(&table_name) {
                //指定名称的表的子事务存在
                table_tr.clone()
            } else {
                //指定名称的表的子事务不存在，则创建指定表的事务，因为是查询操作，所以初始化指定表的子事务为非持久化事务
                self.table_transaction(table_name, table, false, &mut *childes_map)
            };

            match &table_tr {
                KVDBTransaction::RootTr(_tr) => {
                    //忽略键值对数据库的根事务
                    None
                },
                KVDBTransaction::MetaTabTr(tr) => {
                    //获取元信息表的键值对异步流
                    Some(tr.values(key, descending))
                },
                KVDBTransaction::MemOrdTabTr(tr) => {
                    //获取有序内存表的键值对异步流
                    Some(tr.values(key, descending))
                },
                KVDBTransaction::LogOrdTabTr(tr) => {
                    //获取有序日志表的键值对异步流
                    Some(tr.values(key, descending))
                },
                KVDBTransaction::LogWTabTr(tr) => {
                    //获取只写日志表的键值对异步流
                    Some(tr.values(key, descending))
                },
                KVDBTransaction::BtreeOrdTabTr(tr) => {
                    //获取有序B树表的键值对异步流
                    Some(tr.values(key, descending))
                },
            }
        } else {
            //指定名称的表不存在
            None
        }
    }

    /// 锁住指定表的指定关键字
    #[inline]
    async fn lock_key(&self,
                      table_name: Atom,
                      key: Binary) -> Result<(), KVTableTrError> {
        if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_name) {
            //指定名称的表存在，则获取表事务，并开始锁住指定表的指定关键字
            let mut childes_map = self.0.childs_map.lock();
            let table_tr = if let Some(table_tr) = childes_map.get(&table_name) {
                //指定名称的表的子事务存在
                table_tr.clone()
            } else {
                //指定名称的表的子事务不存在，则创建指定表的事务，因为是锁定操作，所以初始化指定表的子事务为非持久化事务
                self.table_transaction(table_name, table, false, &mut *childes_map)
            };

            match &table_tr {
                KVDBTransaction::RootTr(_tr) => {
                    //忽略键值对数据库的根事务
                    Ok(())
                },
                KVDBTransaction::MetaTabTr(tr) => {
                    //锁住元信息表的指定关键字
                    tr.lock_key(key).await
                },
                KVDBTransaction::MemOrdTabTr(tr) => {
                    //锁住有序内存表的指定关键字
                    tr.lock_key(key).await
                },
                KVDBTransaction::LogOrdTabTr(tr) => {
                    //锁住有序日志表的指定关键字
                    tr.lock_key(key).await
                },
                KVDBTransaction::LogWTabTr(tr) => {
                    //锁住只写日志表的指定关键字
                    tr.lock_key(key).await
                },
                KVDBTransaction::BtreeOrdTabTr(tr) => {
                    //锁住有序B树表的指定关键字
                    tr.lock_key(key).await
                },
            }
        } else {
            //指定名称的表不存在
            Ok(())
        }
    }

    /// 解锁指定表的指定关键字
    #[inline]
    async fn unlock_key(&self,
                        table_name: Atom,
                        key: Binary) -> Result<(), KVTableTrError> {
        if let Some(table) = self.0.db_mgr.0.tables.read().await.get(&table_name) {
            //指定名称的表存在，则获取表事务，并开始解锁指定表的指定关键字
            let mut childes_map = self.0.childs_map.lock();
            let table_tr = if let Some(table_tr) = childes_map.get(&table_name) {
                //指定名称的表的子事务存在
                table_tr.clone()
            } else {
                //指定名称的表的子事务不存在，则创建指定表的事务，因为是解锁操作，所以初始化指定表的子事务为非持久化事务
                self.table_transaction(table_name, table, false, &mut *childes_map)
            };

            match &table_tr {
                KVDBTransaction::RootTr(_tr) => {
                    //忽略键值对数据库的根事务
                    Ok(())
                },
                KVDBTransaction::MetaTabTr(tr) => {
                    //解锁元信息表的指定关键字
                    tr.unlock_key(key).await
                },
                KVDBTransaction::MemOrdTabTr(tr) => {
                    //解锁有序内存表的指定关键字
                    tr.unlock_key(key).await
                },
                KVDBTransaction::LogOrdTabTr(tr) => {
                    //解锁有序日志表的指定关键字
                    tr.unlock_key(key).await
                },
                KVDBTransaction::LogWTabTr(tr) => {
                    //解锁只写日志表的指定关键字
                    tr.unlock_key(key).await
                },
                KVDBTransaction::BtreeOrdTabTr(tr) => {
                    //解锁有序B树表的指定关键字
                    tr.unlock_key(key).await
                },
            }
        } else {
            //指定名称的表不存在
            Ok(())
        }
    }

    /// 使用外部读缓存版本和最终写集合，一次性装配并预提交独立版本事务。
    async fn prepare_with_version(&self,
                                  read_set: Vec<TableKeyVersion>,
                                  write_set: Vec<TableKV>)
        -> Result<Vec<u8>, KVTableTrError> {
        if !self.is_writable() {
            return Err(KVTableTrError::new_transaction_error(
                ErrorLevel::Normal,
                "Prepare with version failed, reason: root transaction is read-only"));
        }
        validate_prepare_with_version_inputs(&read_set, &write_set)?;

        // 先在事务私有内存中规范化输入，不触碰表、版本或 manager。分组顺序取两个输入 Vec
        // 第一次出现表的顺序；同一 Table/Key 跨 read/write 重叠时保留 expected 读版本，但最终
        // 动作原子替换为 Write。集合内部重复项已在上方拒绝，避免 HashMap 覆盖掩盖协议错误。
        let mut groups: Vec<VersionPrepareGroup<C, Log>> = Vec::new();
        let mut group_indices: XHashMap<Atom, usize> = XHashMap::default();
        for item in read_set {
            let index = if let Some(index) = group_indices.get(&item.table) {
                *index
            } else {
                let index = groups.len();
                group_indices.insert(item.table.clone(), index);
                groups.push(VersionPrepareGroup {
                    name: item.table.clone(),
                    registered: None,
                    expected: XHashMap::default(),
                    actions: XHashMap::default(),
                    read_keys: Vec::new(),
                    has_write: false,
                });
                index
            };
            let group = &mut groups[index];
            group.read_keys.push(item.key.clone());
            group.expected.insert(item.key.clone(), item.version);
            group.actions.insert(item.key, crate::KVActionLog::Read);
        }
        for item in write_set {
            let index = if let Some(index) = group_indices.get(&item.table) {
                *index
            } else {
                let index = groups.len();
                group_indices.insert(item.table.clone(), index);
                groups.push(VersionPrepareGroup {
                    name: item.table.clone(),
                    registered: None,
                    expected: XHashMap::default(),
                    actions: XHashMap::default(),
                    read_keys: Vec::new(),
                    has_write: false,
                });
                index
            };
            let group = &mut groups[index];
            group.has_write = true;
            group.actions.insert(item.key, crate::KVActionLog::Write(item.value));
        }

        // 这里只克隆注册项并立即释放 registry guard；后续构造、publication await 和 prepare
        // 都不得持有数据库表锁。现存 LogWrite 的 delete 是静态能力错误，必须先于 UID 拒绝。
        {
            let tables = self.0.db_mgr.0.tables.read().await;
            for group in &mut groups {
                group.registered = tables.get(&group.name).cloned();
                if let Some(RegisteredTable {
                    table: KVDBTable::LogWTab(_),
                    ..
                }) = group.registered.as_ref() {
                    if group.actions.values().any(|action| {
                        matches!(action, crate::KVActionLog::Write(None))
                    }) {
                        return Err(KVTableTrError::new_transaction_error(
                            ErrorLevel::Normal,
                            format!("Prepare with version failed, table: {:?}, reason: LogWrite does not support delete",
                                    group.name.as_str())));
                    }
                }
            }
        }

        // 所有子表共享一个只收集“本事务最终写”的回执 owner。它不读取全局最新版本，且只有
        // commit_with_version 会在整棵树提交成功后 take；普通 commit 混用虽被协议禁止，但仍不
        // 能跳过底层版本发布。
        let receipt = VersionReceipt::new();
        let mut identities = Vec::with_capacity(groups.len());
        let mut childes_map = self.0.childs_map.lock();
        for group in groups {
            let identity_versions = group
                .registered
                .as_ref()
                .map(|registered| registered.versions.clone());
            identities.push(RootVersionTableIdentity {
                name: group.name.clone(),
                versions: identity_versions,
                read_keys: group.read_keys,
                has_write: group.has_write,
            });

            if let Some(registered) = group.registered {
                let is_persistent = group.has_write && registered.is_persistent();
                if is_persistent {
                    self.require_persistence();
                }
                self.versioned_table_transaction(group.name,
                                                   &registered,
                                                   is_persistent,
                                                   group.expected,
                                                   receipt.clone(),
                                                   group.actions,
                                                   &mut *childes_map);
            }
        }
        drop(childes_map);
        *self.0.version_context.lock() = Some(RootVersionContext {
            tables: identities,
            receipt,
        });

        self
            .0
            .db_mgr
            .0
            .tr_mgr
            .start(KVDBTransaction::RootTr(self.clone()))
            .await?;
        let prepare_output = self
            .0
            .db_mgr
            .0
            .tr_mgr
            .prepare_all_conflicts(KVDBTransaction::RootTr(self.clone()))
            .await?;

        if self.is_require_persistence() {
            match prepare_output {
                Some(output) if output.len() > 16 => Ok(output),
                _ => Ok(Vec::new()),
            }
        } else {
            Ok(Vec::new())
        }
    }

    /// 异步预提交本次事务对键值对数据库的所有修改，成功返回预提交的输出
    #[inline]
    async fn prepare_modified(&self) -> Result<Vec<u8>, KVTableTrError> {
        if self.get_status() != Transaction2PcStatus::Rollbacked {
            //本次事务的当前状态只要不为回滚成功，则先初始化键值对数据库的根事务
            if let Err(e) = self
                .0
                .db_mgr
                .0
                .tr_mgr
                .start(KVDBTransaction::RootTr(self.clone()))
                .await {
                //初始化键值对数据库的根事务失败，则立即返回错误原因
                return Err(e);
            }
        }

        //预提交键值对数据库的根事务
        match self
            .0
            .db_mgr
            .0
            .tr_mgr
            .prepare(KVDBTransaction::RootTr(self.clone()))
            .await {
            Err(e) => {
                //预提交键值对数据库的根事务失败，则立即返回错误原因
                Err(e)
            },
            Ok(prepare_output) => {
                //预提交键值对数据库的根事务成功
                if self.is_require_persistence() {
                    //本次键值对数据库的根事务，需要持久化
                    if let Some(output) = prepare_output {
                        //键值对数据库的预提交事务，有返回预提交输出
                        if output.len() > 16 {
                            //有效的预提交输出，根事务需要持久化，且至少有一个子事务需要持久化
                            Ok(output)
                        } else {
                            //无效的预提交输出，根事务需要持久化，但所有子事务不需要持久化
                            Ok(vec![])
                        }
                    } else {
                        //预提交键值对数据库的子事务，没有返回预提交输出
                        Ok(vec![])
                    }
                } else {
                    //本次键值对数据库的根事务，不需要持久化
                    Ok(vec![])
                }
            },
        }
    }

    /// 异步预提交本次事务对键值对数据库的所有修改，成功返回预提交的输出，失败返回预提交冲突的首个表名和关键字
    #[inline]
    async fn prepare_modified_conflicts(&self) -> Result<Vec<u8>, KVTableTrError> {
        if self.get_status() != Transaction2PcStatus::Rollbacked {
            //本次事务的当前状态只要不为回滚成功，则先初始化键值对数据库的根事务
            if let Err(e) = self
                .0
                .db_mgr
                .0
                .tr_mgr
                .start(KVDBTransaction::RootTr(self.clone()))
                .await {
                //初始化键值对数据库的根事务失败，则立即返回错误原因
                return Err(e);
            }
        }

        //预提交键值对数据库的根事务
        match self
            .0
            .db_mgr
            .0
            .tr_mgr
            .prepare_conflicts(KVDBTransaction::RootTr(self.clone()))
            .await {
            Err(e) => {
                //预提交键值对数据库的根事务失败，则立即返回错误原因
                Err(e)
            },
            Ok(prepare_output) => {
                //预提交键值对数据库的根事务成功
                if self.is_require_persistence() {
                    //本次键值对数据库的根事务，需要持久化
                    if let Some(output) = prepare_output {
                        //键值对数据库的预提交事务，有返回预提交输出
                        if output.len() > 16 {
                            //有效的预提交输出，根事务需要持久化，且至少有一个子事务需要持久化
                            Ok(output)
                        } else {
                            //无效的预提交输出，根事务需要持久化，但所有子事务不需要持久化
                            Ok(vec![])
                        }
                    } else {
                        //预提交键值对数据库的子事务，没有返回预提交输出
                        Ok(vec![])
                    }
                } else {
                    //本次键值对数据库的根事务，不需要持久化
                    Ok(vec![])
                }
            },
        }
    }

    /// 异步提交本次事务对键值对数据库的所有修改
    #[inline]
    async fn commit_modified(&self, prepare_output: Vec<u8>) -> Result<(), KVTableTrError> {
        self.commit_core(prepare_output).await
    }

    /// 版本提交与普通提交共享唯一 WAL/事务管理器闭环，差异只在成功后的回执所有权。
    async fn commit_with_version(&self,
                                 prepare_output: Vec<u8>)
        -> Result<Vec<TableKeyVersion>, KVTableTrError> {
        let receipt = self
            .0
            .version_context
            .lock()
            .as_ref()
            .expect("Commit with version failed, reason: transaction was not prepared by version protocol")
            .receipt
            .clone();
        match self.commit_core(prepare_output).await {
            Ok(()) => Ok(receipt.take()),
            Err(error) => {
                receipt.clear();
                Err(error)
            },
        }
    }

    async fn commit_core(&self, prepare_output: Vec<u8>) -> Result<(), KVTableTrError> {
        // 为本次事务创建“持久化成功信号”聚合器。计数只包含需要持久化的子表；
        // 根事务自身和非持久化子表不参与。子表数据文件失败时不会调用该回调，根 WAL
        // 因计数未归零而保持未确认。可写事务只有读动作时 prepare_output 和计数都可以为 0：
        // 空输出只跳过 WAL I/O，仍必须由上游 manager 提交整棵 Prepared 树，以释放表级读预留；
        // 此时确认器不会被调用。详见 CONTRACT-CFM-001 和 CONTRACT-TR-005：
        // docs/SEMANTIC_CONTRACTS.md#contract-confirm-success-only。
        // 根 WAL append/flush 的环境/runtime 失败属于 LIMIT-ROOT-WAL-IO-001：上游当前会返回
        // LogCommitFailed，但普通 io::Error 无法证明 WAL 的实际落盘阶段。本层只透传错误，不得
        // 在这里猜测耐久状态、清理 CommitLogger checkpoint 或改变 rollback/Fatal 分类。
        // 当前根 WAL 链最终调用 pi_async_file::AsyncFile::write，而不是未被这四库调用的
        // write_batch；后者的独立实现缺陷归档为 FIND-ASYNC-FILE-BATCH-001，不得据此扩大本边界。
        let commit_confirm = KVDBCommitConfirm::new(self.0.db_mgr.0.rt.clone(),
                                                    self.0.db_mgr.0.tr_mgr.commit_logger(),
                                                    self.get_transaction_uid().unwrap(),
                                                    self.get_commit_uid(),
                                                    self.persistent_children_len());

        //提交键值对数据库的根事务
        match self
            .0
            .db_mgr
            .0
            .tr_mgr
            .commit(KVDBTransaction::RootTr(self.clone()),
                    prepare_output,
                    commit_confirm)
            .await {
            Err(e) => Err(e),
            Ok(_) => {
                //提交键值对数据库的根事务成功，则完成本次键值对数据库事务
                self
                    .0
                    .db_mgr
                    .0
                    .tr_mgr
                    .finish(KVDBTransaction::RootTr(self.clone()));
                Ok(())
            }
        }
    }

    ///
    /// 异步回滚本次事务对键值对数据库的所有修改，事务严重错误无法回滚
    ///
    #[inline]
    async fn rollback_modified(&self) -> Result<(), KVTableTrError> {
        //回滚键值对数据库的根事务
        if let Err(e) = self
            .0
            .db_mgr
            .0
            .tr_mgr
            .rollback(KVDBTransaction::RootTr(self.clone()))
            .await {
            //回滚键值对数据库的根事务失败，则立即返回错误原因
            return Err(e);
        }

        //回滚键值对数据库的根事务成功，则完成本次键值对数据库事务
        self
            .0
            .db_mgr
            .0
            .tr_mgr
            .finish(KVDBTransaction::RootTr(self.clone()));
        Ok(())
    }

    /// 异步预提交本次事务对键值对数据库的所有修复修改，不返回预提交的输出
    async fn prepare_repair(&self,
                            transaction_uid: Guid)
                            -> Result<(), KVTableTrError> {
        let mut childs = self.to_children();
        while let Some(child) = childs.next() {
            match &child {
                KVDBTransaction::MetaTabTr(tr) => {
                    tr.prepare_repair(transaction_uid.clone());
                },
                KVDBTransaction::MemOrdTabTr(tr) => {
                    tr.prepare_repair(transaction_uid.clone());
                },
                KVDBTransaction::LogOrdTabTr(tr) => {
                    tr.prepare_repair(transaction_uid.clone());
                },
                KVDBTransaction::LogWTabTr(tr) => {
                    tr.prepare_repair(transaction_uid.clone());
                },
                KVDBTransaction::BtreeOrdTabTr(tr) => {
                    tr.prepare_repair(transaction_uid.clone());
                },
                KVDBTransaction::RootTr(_) => {
                    //忽略根事务，并继续执行下一个子事务的预提交修复
                    continue;
                }
            }
        }

        Ok(())
    }

    /// 异步提交本次事务对键值对数据库的所有修复修改
    #[inline]
    async fn commit_repair(&self,
                           transaction_uid: Guid,
                           commit_uid: Guid,
                           prepare_output: Vec<u8>) -> Result<(), KVTableTrError> {
        // 恢复提交沿用正常提交的成功信号协议：只有本次重放涉及的全部持久化子表成功，
        // 原根 WAL 才能确认；任一数据文件失败都不发送成功信号。详见 CONTRACT-CFM-001：
        // docs/SEMANTIC_CONTRACTS.md#contract-confirm-success-only。
        let commit_confirm = KVDBCommitConfirm::new(self.0.db_mgr.0.rt.clone(),
                                                    self.0.db_mgr.0.tr_mgr.commit_logger(),
                                                    transaction_uid.clone(),
                                                    Some(commit_uid.clone()),
                                                    self.persistent_children_len());

        //重播提交键值对数据库的根事务
        match self
            .0
            .db_mgr
            .0
            .tr_mgr
            .replay_commit(KVDBTransaction::RootTr(self.clone()),
                           transaction_uid,
                           commit_uid,
                           prepare_output,
                           commit_confirm)
            .await {
            Err(e) => Err(e),
            Ok(_) => {
                //重播提交键值对数据库的根事务成功，则完成本次键值对数据库重播事务
                self
                    .0
                    .db_mgr
                    .0
                    .tr_mgr
                    .finish(KVDBTransaction::RootTr(self.clone()));
                Ok(())
            }
        }
    }
}

/// 版本事务装配时固定的一张表身份和用途。
///
/// `versions` 的 Arc 身份同时代表精确表实例；缺表保存 None。根事务在 Phase 1 和 Phase 2
/// 分别复核一次，防止两阶段之间的 registry 替换被忽略。只读身份失效可枚举 read_keys 为完整
/// 冲突；含写表失效则返回 Common，因为旧表写入不能安全改投到同名新表。
#[derive(Clone)]
struct RootVersionTableIdentity {
    name: Atom,
    versions: Option<KeyVersions>,
    read_keys: Vec<Binary>,
    has_write: bool,
}

/// 一棵独立版本事务树的根级上下文。
///
/// 表身份按首次触表顺序保存；receipt 与所有已创建子事务共享。上下文只属于
/// `prepare_with_version -> commit_with_version` 协议，不得与普通动作、DDL 或普通提交入口混用。
#[derive(Clone)]
struct RootVersionContext {
    tables: Vec<RootVersionTableIdentity>,
    receipt: VersionReceipt,
}

/// `prepare_with_version` 在共享状态副作用前构造的单表规范化输入。
///
/// `expected` 与最终 `actions` 是两个不同维度：跨 read/write 重叠的 Key 同时保留外部读版本和
/// 最终 Write。`registered` 只克隆 registry 中的精确 `(table, versions)` 配对，后续 await 不持
/// tables guard。
struct VersionPrepareGroup<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    name: Atom,
    registered: Option<RegisteredTable<C, Log>>,
    expected: XHashMap<Binary, Version>,
    actions: XHashMap<Binary, crate::KVActionLog>,
    read_keys: Vec<Binary>,
    has_write: bool,
}

///
/// 内部键值对数据库的根事务
///
struct InnerRootTransaction<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    source:             Atom,                                               //事件源
    tid:                SpinLock<Option<Guid>>,                             //事务唯一id
    cid:                SpinLock<Option<Guid>>,                             //事务提交唯一id
    status:             SpinLock<Transaction2PcStatus>,                     //事务状态
    writable:           bool,                                               //事务是否可写
    persistence:        AtomicBool,                                         //事务是否持久化
    prepare_timeout:    u64,                                                //事务预提交超时时长，单位毫秒
    commit_timeout:     u64,                                                //事务提交超时时长，单位毫秒
    childs_map:         SpinLock<XHashMap<Atom, KVDBTransaction<C, Log>>>,  //子事务表
    childs:             SpinLock<KVDBChildTrList<C, Log>>,                  //子事务列表
    db_mgr:             KVDBManager<C, Log>,                                //键值对数据库管理器
    version_context:    SpinLock<Option<RootVersionContext>>,               //版本协议表身份和提交回执
}

/// 数据库注册表中五种物理/逻辑表实现的类型擦除句柄。
///
/// clone 只克隆内部表句柄，不复制数据。应用层从 [`KVDBManager`] 的公开方法操作表；直接
/// 构造 variant 会绕过名称、元数据、版本 registry 和目录注册协议，属于禁止调用域。
#[derive(Clone)]
pub enum KVDBTable<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> {
    /// 记录数据库表定义的 Meta 表。
    MetaTab(MetaTable<C, Log>),
    /// COW 有序 Map 支撑的 Memory 表。
    MemOrdTab(MemoryOrderedTable<C, Log>),
    /// 日志文件支撑的可查询 LogOrdered 表。
    LogOrdTab(LogOrderedTable<C, Log>),
    /// 只写 LogWrite 表；当前不允许外部业务使用。
    LogWTab(LogWriteTable<C, Log>),
    /// 事务写缓存与 redb 数据文件共同支撑的 Btree 表。
    BtreeOrdTab(BtreeOrderedTable<C, Log>),
}

// SAFETY: 所有 variant 的表句柄都由 Arc/线程安全存储和同步原语持有；移动枚举仅移动共享
// owner，不移动自引用对象或暴露可变别名。表级合法调用协议不由该 marker 保证。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Send for KVDBTable<C, Log> {}
// SAFETY: 各表实现负责其 COW root、缓存、日志/redb 句柄的同步；枚举没有额外内部可变性。
unsafe impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> Sync for KVDBTable<C, Log> {}

impl<
    C: Clone + Send + 'static,
    Log: AsyncCommitLog<C = C, Cid = Guid>,
> KVDBTable<C, Log> {
    /// 是否可持久化的表
    #[inline]
    pub fn is_persistent(&self) -> bool {
        match self {
            Self::MetaTab(tab) => tab.is_persistent(),
            Self::MemOrdTab(tab) => tab.is_persistent(),
            Self::LogOrdTab(tab) => tab.is_persistent(),
            Self::LogWTab(tab) => tab.is_persistent(),
            Self::BtreeOrdTab(tab) => tab.is_persistent(),
        }
    }
}

// 数据库跟踪循环
#[cfg(feature = "trace")]
async fn loop_tracing<R, C, Log>(rt: R,
                                 db_mgr: KVDBManager<C, Log>,
                                 interval: usize)
    where R: AsyncRuntime,
          C: Clone + Send + 'static,
          Log: AsyncCommitLog<C = C, Cid = Guid>,
{
    let table_cache_meter = get_table_cache_size_meter(rt.clone())
        .await
        .u64_gauge("pi_db.db.table_cache_size")
        .build();
    loop {
        rt.timeout(interval).await;
        let now = Instant::now();
        for table in db_mgr.tables().await {
            if let Some(size) = db_mgr.table_cache_size(&table).await {
                table_cache_meter.record(size, &[KeyValue::new("table",
                                                               table.as_str().to_string())]);
                rt.timeout(0).await;
            }
        }
        info!("Loop tracing succeeded, interval: {:?}ms, time: {:?}",
            interval,
            now.elapsed());
    }
}

// 将表名序列化为二进制数据
pub(crate) fn table_to_binary(table_name: &Atom) -> Binary {
    let mut buffer = WriteBuffer::new();
    table_name.encode(&mut buffer);
    Binary::new(buffer.bytes)
}

// 将二进制数据反序列化为表名
pub(crate) fn binary_to_table(bin: &Binary) -> Result<Atom, ReadBonErr> {
    let mut buffer = ReadBuffer::new(bin, 0);
    Atom::decode(&mut buffer)
}

// 校验完整表名的 UTF-8 字节长度。调用方选择 InvalidInput（公开 DDL 参数）或 InvalidData
//（Meta/WAL 恢复数据）；错误消息只记录长度，避免把最多数 KiB 的名称复制进日志。
#[inline]
fn validate_table_name(table_name: &Atom,
                       error_kind: ErrorKind,
                       operation: &'static str) -> IOResult<()> {
    let bytes_len = table_name.as_str().as_bytes().len();
    if bytes_len == 0 || bytes_len > MAX_TABLE_NAME_BYTES {
        return Err(Error::new(error_kind,
                              format!("{} failed, table_name_bytes: {}, valid_range: 1..={}, reason: invalid table name length",
                                      operation,
                                      bytes_len,
                                      MAX_TABLE_NAME_BYTES)));
    }

    Ok(())
}

/// 校验版本协议的单个 Table/Key，必须在 Guid 分配、版本写入和事务创建前调用。
#[inline]
fn validate_version_table_key(table: &Atom,
                              key: &Binary,
                              operation: &'static str) -> Result<(), KVTableTrError> {
    validate_table_name(table, ErrorKind::InvalidInput, operation).map_err(|e| {
        KVTableTrError::new_transaction_error(ErrorLevel::Normal, e)
    })?;
    if key.len() == 0 || key.len() > u16::MAX as usize {
        return Err(KVTableTrError::new_transaction_error(
            ErrorLevel::Normal,
            format!("{} failed, table: {:?}, key_bytes: {}, valid_range: 1..={}, reason: invalid key length",
                    operation,
                    table.as_str(),
                    key.len(),
                    u16::MAX)));
    }

    Ok(())
}

/// 静态校验版本事务的两个输入集合；本函数无共享状态副作用，必须先于事务 UID 和子事务创建。
fn validate_prepare_with_version_inputs(read_set: &[TableKeyVersion],
                                        write_set: &[TableKV])
    -> Result<(), KVTableTrError> {
    let mut read_seen: XHashMap<Atom, XHashMap<Binary, ()>> = XHashMap::default();
    for item in read_set {
        validate_version_table_key(&item.table, &item.key, "prepare with version read set")?;
        if read_seen
            .entry(item.table.clone())
            .or_default()
            .insert(item.key.clone(), ())
            .is_some() {
            return Err(KVTableTrError::new_transaction_error(
                ErrorLevel::Normal,
                format!("Prepare with version failed, table: {:?}, key_bytes: {}, reason: duplicate key in read set",
                        item.table.as_str(),
                        item.key.len())));
        }
    }

    let mut write_seen: XHashMap<Atom, XHashMap<Binary, ()>> = XHashMap::default();
    for item in write_set {
        validate_version_table_key(&item.table, &item.key, "prepare with version write set")?;
        if let Some(value) = item.value.as_ref() {
            if value.len() == 0 || value.len() > u32::MAX as usize {
                return Err(KVTableTrError::new_transaction_error(
                    ErrorLevel::Normal,
                    format!("Prepare with version failed, table: {:?}, value_bytes: {}, valid_range: 1..={}, reason: invalid value length",
                            item.table.as_str(),
                            value.len(),
                            u32::MAX)));
            }
        }
        if write_seen
            .entry(item.table.clone())
            .or_default()
            .insert(item.key.clone(), ())
            .is_some() {
            return Err(KVTableTrError::new_transaction_error(
                ErrorLevel::Normal,
                format!("Prepare with version failed, table: {:?}, key_bytes: {}, reason: duplicate key in write set",
                        item.table.as_str(),
                        item.key.len())));
        }
    }

    Ok(())
}

// 删表专用校验：除通用长度契约外，内部 Meta 表永远不能成为公开删除或 WAL 删除恢复目标。
// 调用方必须在根持久化标记、注册表锁和 Meta 动作之前调用，以保证拒绝路径无副作用。
#[inline]
fn validate_removable_table_name(table_name: &Atom,
                                 error_kind: ErrorKind,
                                 operation: &'static str) -> IOResult<()> {
    validate_table_name(table_name, error_kind, operation)?;
    if table_name.as_str() == DEFAULT_DB_TABLES_META_DIR {
        return Err(Error::new(error_kind,
                              format!("{} failed, table_name: {:?}, reason: reserved internal meta table cannot be removed",
                                      operation,
                                      DEFAULT_DB_TABLES_META_DIR)));
    }

    Ok(())
}