use serde::{Deserialize, Serialize};
use sqlparser::dialect::{MsSqlDialect, PostgreSqlDialect};
use sqlparser::tokenizer::{Token, Tokenizer};

use crate::models::connection::DatabaseType;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ExplainFormat {
    Json,
    Standard,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExplainSqlOptions {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub database_type: Option<DatabaseType>,
    /// MySQL supports both a structured JSON plan and the traditional tabular plan.
    /// Omitted formats retain the existing JSON behavior.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub format: Option<ExplainFormat>,
    /// PostgreSQL and SQL Server only: run the statement and report measured
    /// rows/timings. Every other engine ignores the flag.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub analyze: Option<bool>,
    pub sql: String,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExplainSqlBuildResult {
    pub ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sql: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

/// Payload encoding of an estimated plan produced by [`build_explain_sql`].
///
/// Callers that hand a raw plan to a consumer which cannot inspect the SQL
/// (the plugin Host API) need to know how to decode it. Keep this in sync with
/// the generated statements in [`build_explain_sql`];
/// `estimated_plan_format_matches_generated_sql` guards the pairing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EstimatedPlanFormat {
    /// A JSON document returned as a single cell (PostgreSQL, MySQL, OceanBase Oracle).
    Json,
    /// A `ShowPlanXML` document returned as text (SQL Server `SET SHOWPLAN_XML`).
    Xml,
    /// A human-readable plan listing (Dameng, Doris, QuestDB, Oracle).
    Text,
}

impl EstimatedPlanFormat {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Json => "json",
            Self::Xml => "xml",
            Self::Text => "text",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DroppedFilePreviewSqlOptions {
    pub path: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
}

pub fn build_explain_sql(options: ExplainSqlOptions) -> ExplainSqlBuildResult {
    if !supports_explain_plan(options.database_type) {
        return explain_err("unsupported");
    }

    let source = strip_trailing_semicolons(options.sql.trim());
    if source.is_empty() {
        return explain_err("empty");
    }
    if !is_safe_explain_sql_for_database(&source, options.database_type) {
        return explain_err("unsafe");
    }
    if options.analyze == Some(true)
        && options.database_type.is_some_and(|database_type| {
            matches!(database_type, DatabaseType::Postgres | DatabaseType::SqlServer)
                && is_write_sql_for_database(&source, database_type)
        })
    {
        return explain_err("unsafe");
    }
    if options.database_type == Some(DatabaseType::SqlServer) && crate::sql::split_sql_batches(&source).len() != 1 {
        return explain_err("unsafe");
    }

    let sql = match options.database_type {
        // ANALYZE executes the statement; is_safe_explain_sql has already limited
        // the source to SELECT/WITH/TABLE/VALUES. MongoDb shares the plain arm only.
        Some(DatabaseType::Postgres) if options.analyze == Some(true) => {
            format!("EXPLAIN (ANALYZE, FORMAT JSON) {source}")
        }
        Some(DatabaseType::Postgres | DatabaseType::MongoDb) => {
            format!("EXPLAIN (FORMAT JSON) {source}")
        }
        Some(DatabaseType::Dameng | DatabaseType::Questdb) => {
            format!("EXPLAIN {source}")
        }
        Some(DatabaseType::Doris) => format!("EXPLAIN {source}"),
        Some(DatabaseType::Oracle) => format!("EXPLAIN PLAN FOR {source}"),
        // STATISTICS XML returns the same ShowPlanXML document plus per-operator
        // runtime counters, at the price of actually running the statement.
        Some(DatabaseType::SqlServer) if options.analyze == Some(true) => {
            format!("SET STATISTICS XML ON;\nGO\n{source}\nGO\nSET STATISTICS XML OFF;")
        }
        Some(DatabaseType::SqlServer) => {
            format!("SET SHOWPLAN_XML ON;\nGO\n{source}\nGO\nSET SHOWPLAN_XML OFF;")
        }
        Some(DatabaseType::Mysql) if options.format == Some(ExplainFormat::Standard) => {
            // MySQL 8.0.32+ may otherwise inherit TREE or JSON from the session-level explain_format.
            format!("EXPLAIN FORMAT=TRADITIONAL {source}")
        }
        _ => format!("EXPLAIN FORMAT=JSON {source}"),
    };
    ExplainSqlBuildResult { ok: true, sql: Some(sql), reason: None }
}

pub fn build_dropped_file_preview_sql(options: DroppedFilePreviewSqlOptions) -> Option<String> {
    let lower = options.path.to_lowercase();
    let escaped = options.path.replace('\'', "''");
    let limit = options.limit.unwrap_or(1000).max(1);
    if lower.ends_with(".parquet") {
        return Some(format!("SELECT * FROM read_parquet('{escaped}') LIMIT {limit}"));
    }
    if lower.ends_with(".csv") {
        return Some(format!("SELECT * FROM read_csv('{escaped}') LIMIT {limit}"));
    }
    if lower.ends_with(".tsv") {
        return Some(format!("SELECT * FROM read_csv('{escaped}', delim='\\t') LIMIT {limit}"));
    }
    if lower.ends_with(".json") {
        return Some(format!("SELECT * FROM read_json('{escaped}') LIMIT {limit}"));
    }
    None
}

/// How a consumer should decode the estimated plan `build_explain_sql` asks
/// the server for. Dameng and Oracle plans reach DBX through the driver's
/// native plan text rather than a generated `EXPLAIN` statement, and every
/// agent plan is text.
pub fn estimated_plan_format(database_type: Option<DatabaseType>) -> EstimatedPlanFormat {
    match database_type {
        Some(DatabaseType::SqlServer) => EstimatedPlanFormat::Xml,
        Some(DatabaseType::Dameng | DatabaseType::Questdb | DatabaseType::Doris | DatabaseType::Oracle) => {
            EstimatedPlanFormat::Text
        }
        _ => EstimatedPlanFormat::Json,
    }
}

pub fn supports_explain_plan(database_type: Option<DatabaseType>) -> bool {
    matches!(
        database_type,
        Some(
            DatabaseType::Mysql
                | DatabaseType::Doris
                | DatabaseType::Postgres
                | DatabaseType::Questdb
                | DatabaseType::Dameng
                | DatabaseType::Oracle
                | DatabaseType::OceanbaseOracle
                | DatabaseType::SqlServer
        )
    )
}

pub fn is_safe_explain_sql(sql: &str) -> bool {
    is_safe_explain_sql_for_database(sql, None)
}

pub fn is_safe_explain_sql_for_database(sql: &str, database_type: Option<DatabaseType>) -> bool {
    let source = strip_trailing_semicolons(sql.trim());
    if source.is_empty() || has_extra_statement_after_semicolon(&source) {
        return false;
    }
    if database_type == Some(DatabaseType::Oracle) && is_safe_oracle_explain_dml_source(&source) {
        return true;
    }
    is_safe_explain_source(&source) && !contains_dangerous_sql_keyword(&source)
}

/// Returns true for databases that support SQL query execution (execute_query / get_sample_data).
/// Non-SQL databases (Redis, MongoDB, Elasticsearch, InfluxDB, VictoriaMetrics, Neo4j, etcd) are excluded.
pub fn supports_sql_query(database_type: DatabaseType) -> bool {
    !matches!(
        database_type,
        DatabaseType::Redis
            | DatabaseType::MongoDb
            | DatabaseType::Elasticsearch
            | DatabaseType::Easysearch
            | DatabaseType::Solr
            | DatabaseType::Qdrant
            | DatabaseType::Milvus
            | DatabaseType::Weaviate
            | DatabaseType::ChromaDb
            | DatabaseType::InfluxDb
            | DatabaseType::InfluxDb3
            | DatabaseType::VictoriaMetrics
            | DatabaseType::Neo4j
            | DatabaseType::Etcd
    )
}

pub fn is_safe_dameng_autotrace_sql(sql: &str) -> bool {
    let source = strip_trailing_semicolons(sql.trim());
    if source.is_empty() || has_extra_statement_after_semicolon(&source) {
        return false;
    }
    is_safe_explain_source(&source) && !contains_dangerous_sql_keyword(&source)
}

fn explain_err(reason: &str) -> ExplainSqlBuildResult {
    ExplainSqlBuildResult { ok: false, sql: None, reason: Some(reason.to_string()) }
}

fn strip_trailing_semicolons(sql: &str) -> String {
    sql.trim_end().trim_end_matches(';').trim_end().to_string()
}

fn is_safe_explain_source(sql: &str) -> bool {
    let source = strip_sql_comments(sql).trim_start().to_lowercase();
    ["select", "with", "table", "values"].iter().any(|keyword| starts_with_explain_keyword_boundary(&source, keyword))
}

fn starts_with_explain_keyword_boundary(source: &str, keyword: &str) -> bool {
    let Some(remainder) = source.strip_prefix(keyword) else {
        return false;
    };

    match remainder.chars().next() {
        None => true,
        Some(character) if character.is_ascii_alphanumeric() || matches!(character, '_' | '$') => false,
        // The safety gate permits ASCII whitespace and punctuation after the
        // leading keyword, but fails closed for non-ASCII continuation chars.
        Some(character) if character.is_ascii() => true,
        Some(_) => false,
    }
}

fn is_safe_oracle_explain_dml_source(sql: &str) -> bool {
    let source = strip_sql_comments_and_literals(sql);
    let source = source.trim_start().to_ascii_uppercase();
    ["INSERT", "UPDATE", "DELETE", "MERGE"].iter().any(|keyword| starts_with_keyword(&source, keyword))
}

/// Write verbs that mark a read-leading statement as a write, for example a
/// data-modifying CTE (`WITH x AS (DELETE FROM t RETURNING *) SELECT * FROM x`).
const DANGEROUS_SQL_KEYWORDS: &[&str] =
    &["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"];

/// Write verbs that are also scalar function names, so `REPLACE(col, 'a', 'b')`
/// and `INSERT(str, pos, len, new)` are ordinary parts of a read-only query.
///
/// Only these two verbs are exempted when they are called with parentheses.
/// `update` stays dangerous even then: MySQL accepts a parenthesized table
/// reference (`UPDATE (t) SET c = 1`, verified on MySQL 8.4), while
/// `REPLACE (`/`INSERT (` are rejected there as statements, so a parenthesized
/// call can never hide a write for the exempted verbs.
const WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS: &[&str] = &["replace", "insert"];

pub fn contains_dangerous_sql_keyword(sql: &str) -> bool {
    let source = strip_sql_comments_and_literals(sql).to_lowercase();
    DANGEROUS_SQL_KEYWORDS.iter().any(|keyword| contains_dangerous_sql_verb(&source, keyword))
}

/// Match a write verb as a whole word. A verb called with parentheses is a
/// scalar function for [`WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS`] (`SELECT
/// REPLACE(name, 'a', 'b') FROM users` must stay read-only), including when the
/// dialect allows whitespace before the parenthesis.
fn contains_dangerous_sql_verb(source: &str, keyword: &str) -> bool {
    let bytes = source.as_bytes();
    let keyword_bytes = keyword.as_bytes();
    if keyword_bytes.is_empty() || bytes.len() < keyword_bytes.len() {
        return false;
    }
    let allows_function_form = WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS.contains(&keyword);

    for idx in 0..=bytes.len() - keyword_bytes.len() {
        if &bytes[idx..idx + keyword_bytes.len()] != keyword_bytes {
            continue;
        }
        if is_identifier_byte(idx.checked_sub(1).and_then(|i| bytes.get(i)).copied()) {
            continue;
        }
        let rest = &bytes[idx + keyword_bytes.len()..];
        match rest.first().copied() {
            None => return true,
            Some(b'(') if allows_function_form => continue,
            Some(byte) if is_identifier_byte(Some(byte)) => continue,
            Some(byte) => {
                if allows_function_form && byte.is_ascii_whitespace() && starts_with_parenthesis(rest) {
                    continue;
                }
                return true;
            }
        }
    }
    false
}

/// True when the first non-whitespace byte is `(` (SQL Server style `REPLACE (a, b, c)`).
fn starts_with_parenthesis(rest: &[u8]) -> bool {
    let offset = rest.iter().take_while(|byte| byte.is_ascii_whitespace()).count();
    rest.get(offset) == Some(&b'(')
}

/// Keywords that start a read-only SQL statement.
/// Note: FROM is a DuckDB-specific read keyword supporting SELECT-less FROM syntax
/// (e.g. `FROM table SELECT *`). In other databases, a statement starting with FROM
/// is invalid and would be rejected by the database itself, so allowing it poses no risk.
///
/// PRAGMA is intentionally NOT in this list because some PRAGMA statements modify
/// database or session state (e.g. SQLite `PRAGMA journal_mode=WAL`). Instead,
/// read-only PRAGMA forms are handled separately in `is_safe_read_pragma`.
const READ_SQL_KEYWORDS: &[&str] = &["SELECT", "WITH", "SHOW", "DESCRIBE", "DESC", "EXPLAIN", "FROM"];

/// PRAGMA names that are known to be safe read-only queries in SQLite/DuckDB.
/// Only the function-call form `PRAGMA name(args)` matching these names is allowed.
/// Any PRAGMA with assignment (`PRAGMA name = value`) or not in this list is blocked.
const SAFE_READ_PRAGMA_NAMES: &[&str] = &[
    "TABLE_INFO",
    "TABLE_XINFO",
    "INDEX_LIST",
    "INDEX_INFO",
    "FOREIGN_KEY_LIST",
    "DATABASE_LIST",
    "COMPILE_OPTIONS",
    "DATA_VERSION",
];

/// Returns true if the SQL statement is a write operation (not a pure read).
///
/// Callers that know the connection database type should use
/// [`is_write_sql_for_database`] so executable comments are interpreted using
/// the correct dialect. This untyped helper deliberately remains conservative
/// for executable comments.
pub fn is_write_sql(sql: &str) -> bool {
    is_write_sql_with_database_type(sql, None)
}

/// Returns true if the SQL statement is a write operation for a database
/// dialect. In addition to ordinary write statements, this recognizes MySQL
/// executable comments and file exports, plus PostgreSQL-family/SQL Server
/// `SELECT ... INTO` table creation.
pub fn is_write_sql_for_database(sql: &str, database_type: DatabaseType) -> bool {
    // VictoriaMetrics execution is hard-wired to the read-only query API; MetricsQL
    // does not use SQL verbs and must not be rejected by SQL write classification.
    if database_type == DatabaseType::VictoriaMetrics {
        return false;
    }
    if database_type == DatabaseType::DynamoDb {
        if let Some(is_write) = classify_dynamodb_statement_write(sql) {
            return is_write;
        }
    }
    if let Some(risk) = classify_search_engine_query_risk(sql, database_type) {
        return risk != SearchEngineQueryRisk::ReadOnly;
    }
    is_write_sql_with_database_type(sql, Some(database_type))
}

fn classify_dynamodb_statement_write(source: &str) -> Option<bool> {
    let header = source.lines().find(|line| !line.trim().is_empty())?.trim().to_ascii_uppercase();
    match header.as_str() {
        "DBX DYNAMODB SCAN" | "DBX DYNAMODB QUERY / SCAN" => Some(false),
        "DBX DYNAMODB INSERT ITEM" | "DBX DYNAMODB PUT ITEM" | "DBX DYNAMODB DELETE ITEM" => Some(true),
        _ if header.starts_with("DBX DYNAMODB") => Some(true),
        _ => None,
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SearchEngineQueryRisk {
    ReadOnly,
    Write,
    Dangerous,
}

pub fn classify_search_engine_query_risk(source: &str, database_type: DatabaseType) -> Option<SearchEngineQueryRisk> {
    if database_type == DatabaseType::Solr {
        return classify_solr_query_risk(source);
    }
    if !matches!(database_type, DatabaseType::Elasticsearch | DatabaseType::Easysearch) {
        return None;
    }
    let source = strip_leading_search_engine_comments(source);
    let request_line = source.lines().next()?.trim();
    let mut parts = request_line.split_whitespace();
    let method = parts.next()?.to_ascii_uppercase();
    let path = parts.next()?;
    let path = path.split('?').next().unwrap_or(path).trim_end_matches('/');
    let segments = path.split('/').filter(|segment| !segment.is_empty()).collect::<Vec<_>>();
    let has_segment = |candidate: &str| segments.iter().any(|segment| segment.eq_ignore_ascii_case(candidate));
    let has_document_id = |candidate: &str| {
        segments
            .iter()
            .position(|segment| segment.eq_ignore_ascii_case(candidate))
            .is_some_and(|index| segments.get(index + 1).is_some_and(|value| !value.is_empty()))
    };

    match method.as_str() {
        "GET" | "HEAD" | "OPTIONS" => Some(SearchEngineQueryRisk::ReadOnly),
        "POST"
            if [
                "_search",
                "_count",
                "_sql",
                "_msearch",
                "_field_caps",
                "_terms_enum",
                "_validate",
                "_explain",
                "_rank_eval",
                "_search_shards",
            ]
            .iter()
            .any(|endpoint| has_segment(endpoint)) =>
        {
            Some(SearchEngineQueryRisk::ReadOnly)
        }
        "POST" if ["_doc", "_create", "_update", "_bulk"].iter().any(|endpoint| has_segment(endpoint)) => {
            Some(SearchEngineQueryRisk::Write)
        }
        "PUT" if has_document_id("_doc") || has_document_id("_create") => Some(SearchEngineQueryRisk::Write),
        "DELETE" if has_document_id("_doc") => Some(SearchEngineQueryRisk::Write),
        "POST" | "PUT" | "PATCH" | "DELETE" => Some(SearchEngineQueryRisk::Dangerous),
        _ => None,
    }
}

/// Solr REST 写请求集中在 `/{core}/update`（含 JSON/XML/CSV update handlers）；
/// 只读 POST handler 名单之外的写路径（`/admin/*`、`/schema`、`/config`、
/// collection/core 管理）一律按 Dangerous 处理。
fn classify_solr_query_risk(source: &str) -> Option<SearchEngineQueryRisk> {
    const READ_ONLY_POST_ENDPOINTS: &[&str] = &[
        "select",
        "query",
        "get",
        "export",
        "terms",
        "suggest",
        "spell",
        "mlt",
        "sql",
        "graph",
        "clustering",
        "tvrh",
        "luke",
        "elevate",
        "browse",
        "debug",
    ];
    let source = strip_leading_search_engine_comments(source);
    let request_line = source.lines().next()?.trim();
    let mut parts = request_line.split_whitespace();
    let method = parts.next()?.to_ascii_uppercase();
    let raw_path = parts.next()?;
    let path = raw_path.split('?').next().unwrap_or(raw_path).trim_end_matches('/');
    let mut segments: Vec<&str> = path.split('/').filter(|segment| !segment.is_empty()).collect();
    if segments.first().is_some_and(|segment| segment.eq_ignore_ascii_case("solr")) {
        segments.remove(0);
    }
    let endpoint = segments.last().copied().unwrap_or("");
    // update handler 的变体（/update/json、/update/json/docs、/update/csv）在
    // update 段之后还有子路径，所以不能只检查末段。
    let is_update_path = segments.iter().any(|segment| segment.starts_with("update") || *segment == "commit");

    match method.as_str() {
        // Solr 的 update handler 也响应 GET（commit/optimize/stream.body 都能
        // 改数据），所以 update 路径对任何方法都算写，不能只放行 POST。
        _ if is_update_path => Some(SearchEngineQueryRisk::Write),
        "GET" | "HEAD" | "OPTIONS" => {
            if solr_admin_request_is_mutation(raw_path, &segments)
                || solr_replication_request_is_mutation(raw_path, &segments)
            {
                Some(SearchEngineQueryRisk::Dangerous)
            } else {
                Some(SearchEngineQueryRisk::ReadOnly)
            }
        }
        "POST" if READ_ONLY_POST_ENDPOINTS.contains(&endpoint) => Some(SearchEngineQueryRisk::ReadOnly),
        "POST" | "PUT" | "PATCH" | "DELETE" => Some(SearchEngineQueryRisk::Dangerous),
        _ => None,
    }
}

/// CoreAdmin/Collections 管理端点接受 GET 触发的变更动作
///（`GET /admin/cores?action=CREATE&name=x` 会真实建 core），只有显式只读的
/// action 才能算 ReadOnly；其余带 action 的一律按 Dangerous 处理。
fn solr_admin_request_is_mutation(raw_path: &str, segments: &[&str]) -> bool {
    let is_admin_target = segments.windows(2).any(|pair| {
        pair[0].eq_ignore_ascii_case("admin")
            && matches!(pair[1].to_ascii_lowercase().as_str(), "cores" | "collections")
    });
    if !is_admin_target {
        return false;
    }
    let Some(query) = raw_path.split('?').nth(1) else { return false };
    let action = query.split('&').find_map(|param| {
        param.split_once('=').filter(|(key, _)| key.eq_ignore_ascii_case("action")).map(|(_, value)| value)
    });
    match action {
        None => false,
        Some(action) => !matches!(action.to_ascii_uppercase().as_str(), "STATUS" | "REQUESTSTATUS" | "LIST"),
    }
}

/// ReplicationHandler 也接受 GET 触发的变更
///（`GET /{core}/replication?command=disablereplication` 会真实停用复制），
/// 只有显式只读的 command 才算 ReadOnly；不带 command 时 handler 默认返回
/// 详情，按只读处理。
fn solr_replication_request_is_mutation(raw_path: &str, segments: &[&str]) -> bool {
    if !segments.last().is_some_and(|segment| segment.eq_ignore_ascii_case("replication")) {
        return false;
    }
    const READ_ONLY_COMMANDS: &[&str] = &[
        "details",
        "restorestatus",
        "filelist",
        "filecontent",
        "filedownload",
        "filemtime",
        "indexversion",
        "showversion",
    ];
    let Some(query) = raw_path.split('?').nth(1) else {
        return false;
    };
    let command = query.split('&').find_map(|param| {
        param.split_once('=').filter(|(key, _)| key.eq_ignore_ascii_case("command")).map(|(_, value)| value)
    });
    match command {
        None => false,
        Some(command) => !READ_ONLY_COMMANDS.iter().any(|readonly| readonly.eq_ignore_ascii_case(command)),
    }
}

fn strip_leading_search_engine_comments(input: &str) -> &str {
    let mut rest = input;
    loop {
        rest = rest.trim_start();
        if let Some(comment) = rest.strip_prefix('#').or_else(|| rest.strip_prefix("//")) {
            rest = comment.split_once('\n').map_or("", |(_, remaining)| remaining);
            continue;
        }
        if let Some(comment) = rest.strip_prefix("/*") {
            rest = comment.split_once("*/").map_or("", |(_, remaining)| remaining);
            continue;
        }
        return rest.trim();
    }
}

fn is_write_sql_with_database_type(sql: &str, database_type: Option<DatabaseType>) -> bool {
    if database_type == Some(DatabaseType::SqlServer) && is_sqlserver_plan_capture_set(sql) {
        return false;
    }
    if database_type.is_some_and(|database_type| has_dialect_specific_write(sql, database_type)) {
        return true;
    }

    // The untyped helper remains conservative for MySQL executable comments.
    // Typed callers handle those comments in has_dialect_specific_write above.
    let detect_mysql_executable_comments = database_type.is_none();
    let detect_select_into = database_type.is_none();
    let statements = match database_type {
        Some(database_type) => crate::sql::split_sql_statements_for_database(sql, database_type),
        None => crate::sql::split_sql_statements(sql),
    };

    let allow_mysql_checksum = database_type == Some(DatabaseType::Mysql);
    statements.iter().any(|statement| {
        is_write_sql_statement(statement, detect_mysql_executable_comments, detect_select_into, allow_mysql_checksum)
    })
}

/// The explain flows toggle plan capture with a standalone SET statement:
/// `SHOWPLAN_XML` for the estimated plan, `STATISTICS XML` for the actual one.
fn is_sqlserver_plan_capture_set(sql: &str) -> bool {
    let normalized = strip_sql_comments(sql)
        .split_whitespace()
        .map(|part| part.trim_end_matches(';').to_ascii_uppercase())
        .collect::<Vec<_>>();
    let tokens = normalized.iter().map(String::as_str).collect::<Vec<_>>();
    matches!(tokens.as_slice(), ["SET", "SHOWPLAN_XML", "ON" | "OFF"] | ["SET", "STATISTICS", "XML", "ON" | "OFF"])
}

fn is_mysql_compatible_database(database_type: DatabaseType) -> bool {
    matches!(
        database_type,
        DatabaseType::Mysql
            | DatabaseType::Doris
            | DatabaseType::StarRocks
            | DatabaseType::ManticoreSearch
            | DatabaseType::Goldendb
    )
}

fn is_postgresql_family_database(database_type: DatabaseType) -> bool {
    matches!(
        database_type,
        DatabaseType::Postgres
            | DatabaseType::Redshift
            | DatabaseType::Gaussdb
            | DatabaseType::OpenGauss
            | DatabaseType::Kingbase
            | DatabaseType::Highgo
            | DatabaseType::Uxdb
            | DatabaseType::Vastbase
            | DatabaseType::Kwdb
    )
}

/// Detects write-capable syntax that otherwise looks like a read query and is
/// interpreted differently depending on the database dialect.
pub fn has_dialect_specific_write(sql: &str, database_type: DatabaseType) -> bool {
    let statements = crate::sql::split_sql_statements_for_database(sql, database_type);
    statements.iter().any(|statement| has_dialect_specific_write_statement(statement, database_type))
}

fn has_dialect_specific_write_statement(sql: &str, database_type: DatabaseType) -> bool {
    if is_mysql_compatible_database(database_type) {
        let (cleaned, has_executable_comment) = strip_sql_comments_and_literals_with_metadata(sql, true);
        return has_executable_comment
            || contains_keyword_sequence(&cleaned, "INTO", "OUTFILE")
            || contains_keyword_sequence(&cleaned, "INTO", "DUMPFILE");
    }

    if is_postgresql_family_database(database_type) {
        contains_unquoted_keyword(sql, &PostgreSqlDialect {}, "INTO")
    } else {
        match database_type {
            DatabaseType::SqlServer => contains_unquoted_keyword(sql, &MsSqlDialect {}, "INTO"),
            _ => false,
        }
    }
}

fn contains_keyword_sequence(sql: &str, first: &str, second: &str) -> bool {
    let words = sql.split(|ch: char| !ch.is_ascii_alphanumeric() && ch != '_').filter(|word| !word.is_empty());

    let mut previous_matches = false;
    for word in words {
        if previous_matches && word.eq_ignore_ascii_case(second) {
            return true;
        }
        previous_matches = word.eq_ignore_ascii_case(first);
    }
    false
}

pub fn contains_unquoted_keyword(sql: &str, dialect: &dyn sqlparser::dialect::Dialect, keyword: &str) -> bool {
    Tokenizer::new(dialect, sql).tokenize().is_ok_and(|tokens| {
        tokens.into_iter().any(|token| {
            matches!(token, Token::Word(word) if word.quote_style.is_none() && word.value.eq_ignore_ascii_case(keyword))
        })
    })
}

fn is_write_sql_statement(
    sql: &str,
    detect_mysql_executable_comments: bool,
    detect_select_into: bool,
    allow_mysql_checksum: bool,
) -> bool {
    // 1. Strip comments and string literals
    let (cleaned, has_mysql_executable_comment) =
        strip_sql_comments_and_literals_with_metadata(sql, detect_mysql_executable_comments);
    // MySQL/MariaDB executable comments may contain arbitrary SQL, including
    // writes that are not represented by the outer statement (for example,
    // INTO OUTFILE inside a SELECT). Treat them as writes rather than
    // attempting to parse every supported MySQL dialect extension here.
    if has_mysql_executable_comment {
        return true;
    }
    let trimmed = cleaned.trim_start();
    if trimmed.is_empty() {
        return false;
    }
    let upper = trimmed.to_uppercase();

    // 2. Check if first keyword is a read keyword
    let starts_with_read = READ_SQL_KEYWORDS.iter().any(|kw| {
        upper.starts_with(kw) && (upper.len() == kw.len() || !upper.as_bytes()[kw.len()].is_ascii_alphanumeric())
    }) || (allow_mysql_checksum && starts_with_keyword(&upper, "CHECKSUM TABLE"));

    // 3. Special handling for PRAGMA: only allow safe read-only forms
    if !starts_with_read && starts_with_keyword(&upper, "PRAGMA") {
        return !is_safe_read_pragma(&upper);
    }

    // SHOW CREATE returns object metadata; CREATE is part of its read-only syntax.
    if starts_with_show_create(&upper) {
        return false;
    }

    // Untyped callers stay conservative. Typed callers handle SELECT ... INTO
    // only for database families where the syntax performs a write.
    if detect_select_into && starts_with_keyword(&upper, "SELECT") && select_contains_top_level_into(&upper) {
        return true;
    }

    // A statement is a write if it doesn't start with a read keyword,
    // or if it contains embedded dangerous keywords (e.g. CTE-wrapped writes like WITH ... AS (DELETE FROM ...))
    !starts_with_read || contains_dangerous_sql_keyword(sql)
}

fn starts_with_show_create(upper: &str) -> bool {
    let Some(after_show) = upper.strip_prefix("SHOW") else {
        return false;
    };
    if !after_show.is_empty() && after_show.as_bytes()[0].is_ascii_alphanumeric() {
        return false;
    }
    starts_with_keyword(after_show.trim_start(), "CREATE")
}

fn select_contains_top_level_into(upper: &str) -> bool {
    let mut token = String::new();
    let mut depth = 0usize;
    let mut saw_select = false;

    for ch in upper.chars().chain(std::iter::once(' ')) {
        if ch.is_ascii_alphanumeric() || ch == '_' {
            token.push(ch);
            continue;
        }

        if !token.is_empty() {
            if depth == 0 {
                if token == "SELECT" {
                    saw_select = true;
                } else if saw_select && token == "INTO" {
                    return true;
                }
            }
            token.clear();
        }

        if ch == '(' {
            depth += 1;
        } else if ch == ')' {
            depth = depth.saturating_sub(1);
        }
    }

    false
}

/// Check if a PRAGMA statement is a safe read-only form.
/// Allows: PRAGMA table_info(...), PRAGMA index_list(...), etc.
/// Blocks: PRAGMA name = value, PRAGMA name(value), or unknown PRAGMA names.
fn is_safe_read_pragma(upper_stripped: &str) -> bool {
    // Skip "PRAGMA" keyword to get the rest
    let rest = upper_stripped.strip_prefix("PRAGMA").unwrap_or("").trim_start();

    if rest.is_empty() {
        return false;
    }

    // Extract the pragma name (first word)
    let name_end = rest.find(|c: char| !c.is_ascii_alphanumeric() && c != '_').unwrap_or(rest.len());
    let pragma_name = &rest[..name_end];

    // Check if it's in the safe list
    if !SAFE_READ_PRAGMA_NAMES.contains(&pragma_name) {
        return false;
    }

    // Check the form after the name: must be function-call style "(...)" or end of statement
    let after_name = rest[name_end..].trim_start();
    if after_name.is_empty() {
        // PRAGMA table_info (no args) — safe
        return true;
    }
    if after_name.starts_with('(') {
        // PRAGMA table_info(users) — safe read form
        return true;
    }
    // PRAGMA table_info = something or other unsafe form — blocked
    false
}

fn starts_with_keyword(upper: &str, keyword: &str) -> bool {
    upper.starts_with(keyword)
        && (upper.len() == keyword.len() || !upper.as_bytes()[keyword.len()].is_ascii_alphanumeric())
}

/// Whether one MySQL statement is an explicit transaction opener that a
/// tab-scoped auto-commit session may keep alive across executions: a bare
/// `BEGIN` / `BEGIN WORK` / `START TRANSACTION [modifiers]`.
///
/// `COMMIT` / `ROLLBACK` close a transaction instead of opening one, and
/// compound statements (`BEGIN ... END`, only valid inside stored programs)
/// never open a transaction at the top level, so neither is matched. Anything
/// ambiguous — extra statements in the same text, other `BEGIN` continuations —
/// is reported as "not an opener" so the caller falls back to the historical
/// cleanup instead of keeping a transaction open by accident.
pub fn mysql_statement_opens_explicit_transaction(sql: &str) -> bool {
    let cleaned = strip_sql_comments(sql);
    let mut parts = cleaned.split(';');
    let statement = parts.next().unwrap_or_default().trim();
    if statement.is_empty() || parts.any(|part| !part.trim().is_empty()) {
        return false;
    }
    let upper = statement.to_ascii_uppercase();
    let mut tokens = upper.split_whitespace();
    match tokens.next() {
        Some("BEGIN") => matches!(tokens.next(), None | Some("WORK")),
        // Modifiers (`READ ONLY`, `READ WRITE`, `WITH CONSISTENT SNAPSHOT`) are
        // optional and are verified by the server; a malformed one fails the
        // statement, so no transaction is left open for this execution.
        Some("START") => tokens.next() == Some("TRANSACTION"),
        _ => false,
    }
}

/// Check whether a SQL statement is allowed under read-only mode.
/// Returns Err with a descriptive message if the statement is a write operation.
pub fn check_read_only(sql: &str, connection_name: &str, database_type: DatabaseType) -> Result<(), String> {
    if is_write_sql_for_database(sql, database_type) {
        return Err(format!(
            "Read-only mode: connection '{}' has read-only protection enabled. Write operation (including stored procedure calls) blocked.",
            connection_name
        ));
    }
    Ok(())
}

/// Oracle-only, fail-closed classifier for manual-transaction UX. Returns true
/// only for an ordinary top-level read that DBX can prove harmless: a single
/// top-level `SELECT` (not `WITH`) that after Oracle-aware comment/literal
/// stripping contains no row-locking, transaction-control, DDL/DCL, `CALL`/
/// `EXEC`, PL/SQL block, `EXECUTE IMMEDIATE`, database-link, sequence
/// pseudocolumn (`NEXTVAL`/`CURRVAL`), function-invocation, or subquery
/// syntax. Any lexical ambiguity is **not** proven read-only; the classifier
/// deliberately whitelists no functions because user/package function side
/// effects cannot be proven and would create an unbounded compatibility
/// surface.
pub fn is_oracle_proven_read_only_statement(sql: &str) -> bool {
    let cleaned = strip_sql_comments_and_literals(sql);
    let trimmed = cleaned.trim_start();
    if trimmed.is_empty() {
        return false;
    }
    let upper = trimmed.to_uppercase();
    // Single top-level SELECT only (WITH is excluded because data-modifying
    // CTEs can hide writes in a row-returning path).
    if !starts_with_keyword(&upper, "SELECT") {
        return false;
    }

    let lower = cleaned.to_lowercase();

    // Function invocation or subquery syntax: any '(' after literal stripping.
    // This deliberately excludes SELECT upper(name), COUNT(*), package
    // functions, and subqueries DBX cannot prove harmless.
    if lower.contains('(') {
        return false;
    }

    // Database link: '@' cannot appear in an unquoted Oracle identifier, so any
    // remaining '@' after literal stripping is a remote-object reference.
    if lower.contains('@') {
        return false;
    }

    // Word-boundary forbidden keywords. These are Oracle keywords/reserved
    // words, so they cannot be legitimate unquoted identifiers inside a simple
    // read; a quoted occurrence is already stripped to spaces above.
    const FORBIDDEN_WORDS: &[&str] = &[
        // row locking
        "for",
        "update",
        "lock",
        // transaction control
        "begin",
        "commit",
        "rollback",
        "savepoint",
        // DDL/DCL
        "create",
        "alter",
        "drop",
        "truncate",
        "rename",
        "grant",
        "revoke",
        "comment",
        "purge",
        "audit",
        "noaudit",
        "analyze",
        "flashback",
        "merge",
        "insert",
        "delete",
        // set operators (a UNION is not a single top-level SELECT)
        "union",
        "minus",
        "intersect",
        // stored procedures / dynamic SQL / PL/SQL blocks
        "call",
        "exec",
        "execute",
        "declare",
        // sequence pseudocolumns
        "nextval",
        "currval",
        // SELECT ... INTO is PL/SQL-only in Oracle
        "into",
    ];

    !FORBIDDEN_WORDS.iter().any(|word| contains_word(&lower, word))
}

fn contains_word(source: &str, word: &str) -> bool {
    let bytes = source.as_bytes();
    let word_bytes = word.as_bytes();
    if word_bytes.is_empty() || bytes.len() < word_bytes.len() {
        return false;
    }

    for idx in 0..=bytes.len() - word_bytes.len() {
        if &bytes[idx..idx + word_bytes.len()] != word_bytes {
            continue;
        }
        let before = idx.checked_sub(1).and_then(|i| bytes.get(i)).copied();
        let after = bytes.get(idx + word_bytes.len()).copied();
        if !is_identifier_byte(before) && !is_identifier_byte(after) {
            return true;
        }
    }
    false
}

fn is_identifier_byte(byte: Option<u8>) -> bool {
    byte.is_some_and(|b| b.is_ascii_alphanumeric() || b == b'_')
}

fn has_extra_statement_after_semicolon(sql: &str) -> bool {
    let stripped = strip_sql_comments_and_literals(sql);
    stripped.split(';').skip(1).any(|part| !part.trim().is_empty())
}

pub fn strip_sql_comments(sql: &str) -> String {
    let mut output = String::with_capacity(sql.len());
    let mut chars = sql.chars().peekable();
    let mut in_line_comment = false;
    let mut in_block_comment = false;

    while let Some(ch) = chars.next() {
        if in_line_comment {
            if ch == '\n' {
                in_line_comment = false;
                output.push(' ');
            }
            continue;
        }

        if in_block_comment {
            if ch == '*' && chars.peek() == Some(&'/') {
                chars.next();
                in_block_comment = false;
                output.push(' ');
            }
            continue;
        }

        if ch == '-' && chars.peek() == Some(&'-') {
            chars.next();
            in_line_comment = true;
            continue;
        }
        if ch == '#' {
            in_line_comment = true;
            continue;
        }
        if ch == '/' && chars.peek() == Some(&'*') {
            chars.next();
            if let Some(body) = read_mysql_executable_comment_body(&mut chars) {
                output.push(' ');
                output.push_str(&strip_sql_comments(&body));
                output.push(' ');
            } else {
                in_block_comment = true;
            }
            continue;
        }

        output.push(ch);
    }

    output
}

pub fn strip_sql_comments_and_literals(sql: &str) -> String {
    strip_sql_comments_and_literals_with_metadata(sql, false).0
}

fn strip_sql_comments_and_literals_with_metadata(sql: &str, detect_mysql_executable_comments: bool) -> (String, bool) {
    let mut output = String::with_capacity(sql.len());
    let mut chars = sql.chars().peekable();
    let mut in_line_comment = false;
    let mut in_block_comment = false;
    let mut in_single_quote = false;
    let mut in_double_quote = false;
    let mut in_backtick_quote = false;
    let mut has_mysql_executable_comment = false;

    while let Some(ch) = chars.next() {
        if in_line_comment {
            if ch == '\n' {
                in_line_comment = false;
                output.push(' ');
            }
            continue;
        }

        if in_block_comment {
            if ch == '*' && chars.peek() == Some(&'/') {
                chars.next();
                in_block_comment = false;
                output.push(' ');
            }
            continue;
        }

        if in_single_quote {
            if ch == '\'' {
                if chars.peek() == Some(&'\'') {
                    chars.next();
                } else {
                    in_single_quote = false;
                }
            }
            output.push(' ');
            continue;
        }

        if in_double_quote {
            if ch == '"' {
                if chars.peek() == Some(&'"') {
                    chars.next();
                } else {
                    in_double_quote = false;
                }
            }
            output.push(' ');
            continue;
        }

        if in_backtick_quote {
            if ch == '`' {
                if chars.peek() == Some(&'`') {
                    chars.next();
                } else {
                    in_backtick_quote = false;
                }
            }
            output.push(' ');
            continue;
        }

        if ch == '-' && chars.peek() == Some(&'-') {
            chars.next();
            in_line_comment = true;
            continue;
        }
        if ch == '#' {
            in_line_comment = true;
            continue;
        }
        if ch == '/' && chars.peek() == Some(&'*') {
            chars.next();
            if detect_mysql_executable_comments && is_mysql_executable_comment_start(&chars) {
                has_mysql_executable_comment = true;
            }
            in_block_comment = true;
            continue;
        }
        if ch == '\'' {
            in_single_quote = true;
            output.push(' ');
            continue;
        }
        if ch == '"' {
            in_double_quote = true;
            output.push(' ');
            continue;
        }
        if ch == '`' {
            in_backtick_quote = true;
            output.push(' ');
            continue;
        }

        output.push(ch);
    }

    (output, has_mysql_executable_comment)
}

/// `/*! ... */` is executable in MySQL, while `/*M! ... */` (optionally
/// followed by a version number) is executable in MariaDB.
fn is_mysql_executable_comment_start(chars: &std::iter::Peekable<std::str::Chars<'_>>) -> bool {
    let mut marker = chars.clone();
    match marker.next() {
        Some('!') => true,
        Some('M') => marker.next() == Some('!'),
        _ => false,
    }
}

fn read_mysql_executable_comment_body<I>(chars: &mut std::iter::Peekable<I>) -> Option<String>
where
    I: Iterator<Item = char>,
{
    let marker = chars.peek().copied()?;
    if marker == '!' {
        chars.next();
    } else if marker == 'M' {
        chars.next();
        if chars.peek() != Some(&'!') {
            return None;
        }
        chars.next();
    } else {
        return None;
    }

    let mut body = String::new();
    let mut skipping_version = true;
    while let Some(ch) = chars.next() {
        if ch == '*' && chars.peek() == Some(&'/') {
            chars.next();
            return Some(body);
        }
        if skipping_version && (ch.is_ascii_digit() || ch.is_whitespace()) {
            continue;
        }
        skipping_version = false;
        body.push(ch);
    }
    Some(body)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn oracle_proven_read_only_accepts_simple_top_level_selects() {
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM table"));
        assert!(is_oracle_proven_read_only_statement("select id, name from users"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users WHERE id = 1"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users ORDER BY id"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users FETCH FIRST 10 ROWS ONLY"));
        assert!(is_oracle_proven_read_only_statement("SELECT 1"));
        assert!(is_oracle_proven_read_only_statement("SELECT ROWNUM FROM dual"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users"));
        // Comment/literal stripping hides keywords that are not part of the SQL.
        assert!(is_oracle_proven_read_only_statement("-- comment\nSELECT * FROM users"));
        assert!(is_oracle_proven_read_only_statement("SELECT 'update' FROM dual"));
        assert!(is_oracle_proven_read_only_statement("SELECT \"select\" FROM t"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM user_update"));
    }

    #[test]
    fn oracle_proven_read_only_rejects_writes_and_ddl() {
        for sql in [
            "UPDATE users SET name = 'x'",
            "INSERT INTO users VALUES (1)",
            "DELETE FROM users",
            "MERGE INTO target USING source ON (a = b) WHEN MATCHED THEN UPDATE",
            "CREATE TABLE t (id INT)",
            "ALTER TABLE t ADD COLUMN c INT",
            "DROP TABLE t",
            "TRUNCATE TABLE t",
            "GRANT SELECT ON t TO x",
            "REVOKE SELECT ON t FROM x",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_cte_procedure_and_plsql() {
        for sql in [
            "WITH cte AS (SELECT 1 FROM dual) SELECT * FROM cte",
            "CALL my_proc()",
            "EXEC my_proc",
            "EXECUTE IMMEDIATE 'DROP TABLE t'",
            "BEGIN UPDATE t SET x = 1; END;",
            "DECLARE x INT; BEGIN NULL; END;",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_locking_and_sequence_pseudocolumns() {
        for sql in [
            "SELECT * FROM t FOR UPDATE",
            "LOCK TABLE t IN EXCLUSIVE MODE",
            "SELECT seq.NEXTVAL FROM dual",
            "SELECT seq.CURRVAL FROM dual",
            "SELECT * FROM t@dblink",
            "SELECT * FROM t WHERE id = 1 FOR UPDATE",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_functions_and_subqueries() {
        for sql in [
            "SELECT upper(name) FROM users",
            "SELECT count(*) FROM users",
            "SELECT pkg.fn() FROM dual",
            "SELECT * FROM users WHERE id IN (SELECT id FROM orders)",
            "SELECT * FROM (SELECT * FROM users)",
            "SELECT NVL(name, 'x') FROM users",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_empty_and_ambiguous_input() {
        assert!(!is_oracle_proven_read_only_statement(""));
        assert!(!is_oracle_proven_read_only_statement("   "));
        assert!(!is_oracle_proven_read_only_statement("-- only a comment"));
        assert!(!is_oracle_proven_read_only_statement("SELECT * FROM t WHERE a IN (1, 2)"));
        assert!(!is_oracle_proven_read_only_statement("SELECT * FROM t WHERE b = 'x' AND c IN (3)"));
        assert!(!is_oracle_proven_read_only_statement("SELECT 1 UNION SELECT 2"));
    }

    #[test]
    fn builds_postgres_json_explain_sql() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Postgres),
            format: None,
            analyze: None,
            sql: " select * from users where id = 1; ".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN (FORMAT JSON) select * from users where id = 1".to_string()),
                reason: None,
            }
        );
    }

    #[test]
    fn builds_postgres_explain_sql_for_keyword_boundaries() {
        for sql in [
            "SELECT* FROM users",
            "SELECT\t* FROM users",
            "SELECT(1)",
            "VALUES(1)",
            "SELECT\r\n*\r\nFROM users",
            "TABLE users",
        ] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Postgres),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: true, sql: Some(format!("EXPLAIN (FORMAT JSON) {sql}")), reason: None },
                "expected safe EXPLAIN source: {sql:?}"
            );
        }
    }

    #[test]
    fn rejects_explain_keyword_identifier_prefixes() {
        for sql in ["SELECTED", "SELECT_foo", "SELECT$foo", "SELECT1", "VALUES_foo", "VALUES1"] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Postgres),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "expected unsafe EXPLAIN source: {sql:?}"
            );
        }
    }

    #[test]
    fn builds_postgres_analyze_explain_sql() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Postgres),
            format: None,
            analyze: Some(true),
            sql: " select * from users where id = 1; ".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN (ANALYZE, FORMAT JSON) select * from users where id = 1".to_string()),
                reason: None,
            }
        );

        // analyze: Some(false) must behave exactly like the omitted flag.
        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Postgres),
                format: None,
                analyze: Some(false),
                sql: "SELECT 1".to_string(),
            })
            .sql,
            Some("EXPLAIN (FORMAT JSON) SELECT 1".to_string())
        );
    }

    #[test]
    fn builds_sqlserver_actual_plan_explain_sql() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::SqlServer),
            format: None,
            analyze: Some(true),
            sql: " select * from dbo.orders where id = 1; ".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some(
                    "SET STATISTICS XML ON;\nGO\nselect * from dbo.orders where id = 1\nGO\nSET STATISTICS XML OFF;"
                        .to_string()
                ),
                reason: None,
            }
        );

        // analyze: Some(false) must keep the estimated SHOWPLAN plan.
        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::SqlServer),
                format: None,
                analyze: Some(false),
                sql: "SELECT 1".to_string(),
            })
            .sql,
            Some("SET SHOWPLAN_XML ON;\nGO\nSELECT 1\nGO\nSET SHOWPLAN_XML OFF;".to_string())
        );
    }

    #[test]
    fn refuses_sqlserver_analyze_on_non_select_sources() {
        // STATISTICS XML runs the statement, so the safety gate stays load-bearing.
        for sql in [
            "delete from dbo.orders",
            "update dbo.orders set name = 'x'",
            "SELECT * INTO dbo.orders_copy FROM dbo.orders",
            "SELECT 1\nGO\nDROP TABLE dbo.orders",
        ] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::SqlServer),
                    format: None,
                    analyze: Some(true),
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "{sql}"
            );
        }
    }

    #[test]
    fn ignores_analyze_for_every_engine_but_postgres_and_sqlserver() {
        for (db_type, expected) in [
            (DatabaseType::MongoDb, "EXPLAIN (FORMAT JSON) SELECT 1"),
            (DatabaseType::Mysql, "EXPLAIN FORMAT=JSON SELECT 1"),
            (DatabaseType::Dameng, "EXPLAIN SELECT 1"),
            (DatabaseType::Questdb, "EXPLAIN SELECT 1"),
            (DatabaseType::Oracle, "EXPLAIN PLAN FOR SELECT 1"),
        ] {
            let analyzed = build_explain_sql(ExplainSqlOptions {
                database_type: Some(db_type),
                format: None,
                analyze: Some(true),
                sql: "SELECT 1".to_string(),
            });
            let plain = build_explain_sql(ExplainSqlOptions {
                database_type: Some(db_type),
                format: None,
                analyze: None,
                sql: "SELECT 1".to_string(),
            });

            assert_eq!(analyzed, plain, "{db_type:?} must ignore the analyze flag");
            // MongoDb is rejected earlier by supports_explain_plan, so it never
            // reaches the match arm it nominally shares with Postgres.
            if db_type != DatabaseType::MongoDb {
                assert_eq!(analyzed.sql, Some(expected.to_string()), "{db_type:?}");
            } else {
                assert_eq!(analyzed.reason, Some("unsupported".to_string()));
            }
        }
    }

    #[test]
    fn refuses_analyze_on_non_select_sources() {
        for sql in ["delete from users", "update users set name = 'x'", "SELECT 1; DROP TABLE users"] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Postgres),
                    format: None,
                    analyze: Some(true),
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "{sql}"
            );
        }
    }

    #[test]
    fn builds_dameng_explain_sql() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Dameng),
            format: None,
            analyze: None,
            sql: "SELECT * FROM t1 WHERE id = 1".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN SELECT * FROM t1 WHERE id = 1".to_string()),
                reason: None,
            }
        );
    }

    #[test]
    fn builds_doris_explain_sql_without_mysql_json_format() {
        assert!(supports_explain_plan(Some(DatabaseType::Doris)));

        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Doris),
                format: None,
                analyze: None,
                sql: "SELECT * FROM users;".to_string(),
            }),
            ExplainSqlBuildResult { ok: true, sql: Some("EXPLAIN SELECT * FROM users".to_string()), reason: None }
        );

        for sql in ["DELETE FROM users", "SELECT * FROM users; DROP TABLE users"] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Doris),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "{sql}"
            );
        }
    }

    #[test]
    fn builds_oracle_explain_plan_sql() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Oracle),
            format: None,
            analyze: None,
            sql: "WITH rows AS (SELECT 1 AS id FROM dual) SELECT * FROM rows;".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN PLAN FOR WITH rows AS (SELECT 1 AS id FROM dual) SELECT * FROM rows".to_string()),
                reason: None,
            }
        );
    }

    #[test]
    fn builds_oracle_explain_plan_for_update() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Oracle),
            format: None,
            analyze: None,
            sql: "UPDATE AA_PAY_VOUCHER_TEMP t SET t.AGENCY_ID = '1';".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN PLAN FOR UPDATE AA_PAY_VOUCHER_TEMP t SET t.AGENCY_ID = '1'".to_string()),
                reason: None,
            }
        );
    }

    #[test]
    fn allows_oracle_explain_plan_for_dml_only() {
        for (sql, expected) in [
            ("INSERT INTO audit_log (id) VALUES (1)", "EXPLAIN PLAN FOR INSERT INTO audit_log (id) VALUES (1)"),
            ("DELETE FROM audit_log WHERE id = 1", "EXPLAIN PLAN FOR DELETE FROM audit_log WHERE id = 1"),
            (
                "MERGE INTO target t USING source s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.name = s.name",
                "EXPLAIN PLAN FOR MERGE INTO target t USING source s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.name = s.name",
            ),
        ] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Oracle),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                })
                .sql,
                Some(expected.to_string()),
                "{sql}"
            );
        }

        for sql in [
            "DROP TABLE audit_log",
            "ALTER TABLE audit_log ADD created_at DATE",
            "BEGIN DELETE FROM audit_log; END;",
            "UPDATE audit_log SET id = 2; DROP TABLE audit_log",
        ] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Oracle),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "{sql}"
            );
        }
    }

    #[test]
    fn estimated_plan_format_matches_generated_sql() {
        // Every dialect `supports_explain_plan` advertises must declare a payload
        // format that the statement `build_explain_sql` generates really produces.
        for database_type in
            DatabaseType::ALL.iter().copied().filter(|database_type| supports_explain_plan(Some(*database_type)))
        {
            let explain_sql = build_explain_sql(ExplainSqlOptions {
                database_type: Some(database_type),
                format: None,
                analyze: None,
                sql: "SELECT 1".to_string(),
            })
            .sql
            .expect("a supported dialect always builds EXPLAIN SQL");

            match estimated_plan_format(Some(database_type)) {
                EstimatedPlanFormat::Json => assert!(
                    explain_sql.contains("FORMAT JSON") || explain_sql.contains("FORMAT=JSON"),
                    "{database_type:?} declares a JSON plan but generated '{explain_sql}'"
                ),
                EstimatedPlanFormat::Xml => assert!(
                    explain_sql.contains("SHOWPLAN_XML"),
                    "{database_type:?} declares an XML plan but generated '{explain_sql}'"
                ),
                EstimatedPlanFormat::Text => assert!(
                    explain_sql.starts_with("EXPLAIN ") && !explain_sql.contains("FORMAT"),
                    "{database_type:?} declares a text plan but generated '{explain_sql}'"
                ),
            }
        }

        assert_eq!(estimated_plan_format(None), EstimatedPlanFormat::Json);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Dameng)), EstimatedPlanFormat::Text);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Questdb)), EstimatedPlanFormat::Text);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Doris)), EstimatedPlanFormat::Text);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Oracle)), EstimatedPlanFormat::Text);
        assert_eq!(estimated_plan_format(Some(DatabaseType::SqlServer)), EstimatedPlanFormat::Xml);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Mysql)), EstimatedPlanFormat::Json);
    }

    #[test]
    fn builds_sqlserver_showplan_xml_batches() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::SqlServer),
            format: None,
            analyze: None,
            sql: "WITH rows AS (SELECT 1 AS id) SELECT * FROM rows;".to_string(),
        });

        assert_eq!(
            result,
            ExplainSqlBuildResult {
                ok: true,
                sql: Some(
                    "SET SHOWPLAN_XML ON;\nGO\nWITH rows AS (SELECT 1 AS id) SELECT * FROM rows\nGO\nSET SHOWPLAN_XML OFF;"
                        .to_string()
                ),
                reason: None,
            }
        );

        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::SqlServer),
                format: None,
                analyze: None,
                sql: "SELECT 1\nGO\nSELECT 2".to_string(),
            }),
            ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
        );
        assert!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::SqlServer),
                format: None,
                analyze: None,
                sql: "SELECT 'first line\nGO\nlast line' AS text".to_string(),
            })
            .ok
        );
    }

    #[test]
    fn validates_dameng_autotrace_sql_safety() {
        assert!(is_safe_dameng_autotrace_sql("SELECT * FROM t WHERE name = 'delete';"));
        assert!(is_safe_dameng_autotrace_sql("/* comment */ WITH q AS (SELECT 1) SELECT * FROM q"));
        assert!(!is_safe_dameng_autotrace_sql("SELECT * FROM t; DELETE FROM t"));
        assert!(!is_safe_dameng_autotrace_sql("UPDATE t SET name = 'x'"));
        assert!(!is_safe_dameng_autotrace_sql("SELECT * FROM t; /* hidden */ DROP TABLE t"));
        assert!(!is_safe_dameng_autotrace_sql(""));
    }

    #[test]
    fn builds_mysql_json_explain_and_rejects_unsafe_sql() {
        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Mysql),
                format: None,
                analyze: None,
                sql: "SELECT * FROM users;".to_string(),
            }),
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN FORMAT=JSON SELECT * FROM users".to_string()),
                reason: None,
            }
        );

        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Mysql),
                format: None,
                analyze: None,
                sql: "delete from users".to_string(),
            }),
            ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
        );

        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Mysql),
                format: None,
                analyze: None,
                sql: "SELECT * FROM users; DELETE FROM users".to_string(),
            }),
            ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
        );

        assert_eq!(
            build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Mysql),
                format: Some(ExplainFormat::Standard),
                analyze: None,
                sql: "SELECT * FROM users;".to_string(),
            }),
            ExplainSqlBuildResult {
                ok: true,
                sql: Some("EXPLAIN FORMAT=TRADITIONAL SELECT * FROM users".to_string()),
                reason: None,
            }
        );
    }

    #[test]
    fn builds_nonexecuting_oceanbase_oracle_json_explain() {
        let result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::OceanbaseOracle),
            format: Some(ExplainFormat::Json),
            analyze: Some(true),
            sql: "SELECT * FROM events;".to_string(),
        });
        assert_eq!(result.sql.as_deref(), Some("EXPLAIN FORMAT=JSON SELECT * FROM events"));
        assert!(result.ok);

        let unsafe_result = build_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::OceanbaseOracle),
            format: None,
            analyze: None,
            sql: "DELETE FROM events".to_string(),
        });
        assert_eq!(unsafe_result.reason.as_deref(), Some("unsafe"));
    }

    #[test]
    fn builds_dropped_file_preview_sql() {
        assert_eq!(
            build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
                path: "/tmp/O'Hara.csv".to_string(),
                limit: Some(25),
            }),
            Some("SELECT * FROM read_csv('/tmp/O''Hara.csv') LIMIT 25".to_string())
        );
        assert_eq!(
            build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
                path: "/tmp/data.tsv".to_string(),
                limit: None,
            }),
            Some("SELECT * FROM read_csv('/tmp/data.tsv', delim='\\t') LIMIT 1000".to_string())
        );
        assert_eq!(
            build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
                path: "/tmp/data.txt".to_string(),
                limit: None,
            }),
            None
        );
    }

    #[test]
    fn strip_sql_comments_and_literals_basic() {
        assert_eq!(strip_sql_comments_and_literals("SELECT 1"), "SELECT 1");
        assert_eq!(strip_sql_comments_and_literals("SELECT 'hello'"), "SELECT        ");
        assert_eq!(strip_sql_comments_and_literals("SELECT \"hello\""), "SELECT        ");
        assert_eq!(strip_sql_comments_and_literals("-- comment\nSELECT 1"), " SELECT 1");
        assert_eq!(strip_sql_comments_and_literals("/* block */ SELECT 1"), "  SELECT 1");
        assert_eq!(strip_sql_comments_and_literals("# comment\nSELECT 1"), " SELECT 1");
    }

    #[test]
    fn strip_sql_comments_and_literals_nested() {
        // String literals containing comments should be stripped
        assert_eq!(strip_sql_comments_and_literals("SELECT '/* not a comment */'"), "SELECT                      ");
        // Comments containing string delimiters should be stripped
        assert_eq!(strip_sql_comments_and_literals("/* 'not a string' */ SELECT 1"), "  SELECT 1");
    }

    #[test]
    fn contains_dangerous_sql_keyword_detects_writes() {
        assert!(contains_dangerous_sql_keyword("DROP TABLE users"));
        assert!(contains_dangerous_sql_keyword("DELETE FROM users"));
        assert!(contains_dangerous_sql_keyword("TRUNCATE TABLE users"));
        assert!(contains_dangerous_sql_keyword("ALTER TABLE users ADD COLUMN age INT"));
        assert!(contains_dangerous_sql_keyword("UPDATE users SET name = 'x'"));
        assert!(contains_dangerous_sql_keyword("MERGE INTO target USING source"));
        assert!(contains_dangerous_sql_keyword("REPLACE INTO users VALUES (1)"));
        assert!(contains_dangerous_sql_keyword("INSERT INTO users VALUES (1)"));
        assert!(contains_dangerous_sql_keyword("CREATE TABLE users (id INT)"));
    }

    #[test]
    fn contains_dangerous_sql_keyword_ignores_substrings() {
        // "updateable" contains "update" as substring but is a different word
        assert!(!contains_dangerous_sql_keyword("SELECT * FROM updateable_view"));
        // "dropped" contains "drop" as substring
        assert!(!contains_dangerous_sql_keyword("SELECT dropped FROM t"));
        // "inserted" contains "insert"
        assert!(!contains_dangerous_sql_keyword("SELECT inserted FROM t"));
    }

    #[test]
    fn contains_dangerous_sql_keyword_ignores_scalar_functions_named_like_writes() {
        // #9978: REPLACE()/INSERT() are scalar functions, not statements.
        assert!(!contains_dangerous_sql_keyword("SELECT REPLACE(name, 'a', 'b') FROM users"));
        assert!(!contains_dangerous_sql_keyword("SELECT REPLACE (name, 'a', 'b') FROM users"));
        assert!(!contains_dangerous_sql_keyword("SELECT INSERT('Quadratic', 3, 4, 'What')"));
        assert!(!contains_dangerous_sql_keyword("SELECT dropped, inserted FROM t"));
        assert!(contains_dangerous_sql_keyword("REPLACE INTO users VALUES (1)"));
        assert!(contains_dangerous_sql_keyword("REPLACE users SET name = 'x'"));
        assert!(contains_dangerous_sql_keyword("INSERT INTO users VALUES (1)"));
        // `UPDATE (t)` is a real write: MySQL accepts a parenthesized table reference.
        assert!(contains_dangerous_sql_keyword("UPDATE (users) SET name = 'x'"));
    }

    #[test]
    fn is_write_sql_allows_mysql_json_table_query_with_replace_function() {
        // #9978: a JSON_TABLE query using REPLACE() must not be treated as a write,
        // otherwise read-only connections ask for write permission before running it.
        let sql = "with tmp as (\n  select '张三,李四,王五' as `names`\n)\nSELECT jt.name\nFROM tmp t\nJOIN JSON_TABLE(\n    CONCAT('[\"',REPLACE(t.`names`,',','\",\"'),'\"]'),\n    '$[*]' COLUMNS(name VARCHAR(50) PATH '$')\n) jt";
        assert!(!is_write_sql_for_database(sql, DatabaseType::Mysql), "JSON_TABLE + REPLACE() must stay read-only");
        assert!(!is_write_sql_for_database("SELECT REPLACE(name, 'a', 'b') FROM users", DatabaseType::Mysql));
        assert!(!is_write_sql_for_database("SELECT INSERT('Quadratic', 3, 4, 'What')", DatabaseType::Mysql));
        assert!(!is_write_sql_for_database("SELECT REPLACE (name, 'a', 'b') FROM users", DatabaseType::SqlServer));
        assert!(!is_write_sql_for_database("SELECT replace(name, 'a', 'b') FROM users", DatabaseType::Postgres));
        assert!(!is_write_sql(sql));
    }

    #[test]
    fn is_write_sql_still_blocks_writes_using_function_like_verbs() {
        assert!(is_write_sql_for_database("REPLACE INTO users VALUES (1)", DatabaseType::Mysql));
        assert!(is_write_sql_for_database("REPLACE users SET name = 'x'", DatabaseType::Mysql));
        assert!(is_write_sql_for_database("INSERT INTO users VALUES (1)", DatabaseType::Mysql));
        // MySQL accepts a parenthesized table reference, so `UPDATE (` is a write.
        assert!(is_write_sql_for_database("UPDATE (users) SET name = 'x'", DatabaseType::Mysql));
        assert!(is_write_sql_for_database(
            "WITH x AS (SELECT 1 AS n) UPDATE (users) SET name = 'x'",
            DatabaseType::Mysql
        ));
        // Data-modifying CTEs keep being detected.
        assert!(is_write_sql_for_database(
            "WITH x AS (INSERT INTO users VALUES (1) RETURNING *) SELECT * FROM x",
            DatabaseType::Postgres
        ));
        assert!(is_write_sql_for_database(
            "WITH x AS (DELETE FROM users RETURNING *) SELECT * FROM x",
            DatabaseType::Postgres
        ));
    }

    #[test]
    fn contains_dangerous_sql_keyword_ignores_in_string_literals() {
        assert!(!contains_dangerous_sql_keyword("SELECT 'DROP TABLE users' FROM t"));
        assert!(!contains_dangerous_sql_keyword("SELECT 'delete' FROM t"));
        assert!(!contains_dangerous_sql_keyword("SELECT \"CREATE TABLE\" FROM t"));
    }

    #[test]
    fn contains_dangerous_sql_keyword_ignores_backtick_identifiers() {
        for keyword in ["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"] {
            let sql = format!("SELECT 1 AS `{keyword}`");
            assert!(!contains_dangerous_sql_keyword(&sql), "expected safe SQL: {sql}");
        }
        assert!(!contains_dangerous_sql_keyword("SELECT 1 AS `before``delete`"));
    }

    #[test]
    fn is_write_sql_detects_simple_writes() {
        assert!(is_write_sql("INSERT INTO users VALUES (1)"));
        assert!(is_write_sql("UPDATE users SET name = 'x'"));
        assert!(is_write_sql("DELETE FROM users"));
        assert!(is_write_sql("DROP TABLE users"));
        assert!(is_write_sql("CREATE TABLE users (id INT)"));
        assert!(is_write_sql("ALTER TABLE users ADD COLUMN age INT"));
        assert!(is_write_sql("TRUNCATE TABLE users"));
        assert!(is_write_sql("EXPLAIN ANALYZE DELETE FROM users"));
        assert!(is_write_sql("SELECT * INTO backup_users FROM users"));
        assert!(is_write_sql("SELECT * FROM users INTO OUTFILE '/tmp/users.csv'"));
        assert!(is_write_sql("COPY users FROM '/tmp/users.csv'"));
        assert!(is_write_sql("/*! DELETE FROM users */"));
        assert!(is_write_sql(
            "MERGE INTO target USING source ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.name = s.name"
        ));
        assert!(is_write_sql("REPLACE INTO users VALUES (1)"));
    }

    #[test]
    fn is_write_sql_allows_reads() {
        assert!(!is_write_sql("SELECT * FROM users"));
        assert!(!is_write_sql("SELECT id, name FROM users WHERE active = true"));
        assert!(!is_write_sql("WITH cte AS (SELECT 1) SELECT * FROM cte"));
        assert!(!is_write_sql("SHOW TABLES"));
        assert!(!is_write_sql("DESCRIBE users"));
        assert!(!is_write_sql("DESC users"));
        assert!(!is_write_sql("EXPLAIN SELECT * FROM users"));
        assert!(!is_write_sql("PRAGMA table_info(users)"));
        assert!(!is_write_sql("FROM users SELECT *"));
    }

    #[test]
    fn is_write_sql_allows_backtick_identifiers_with_dangerous_keywords() {
        for keyword in ["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"] {
            let sql = format!("SELECT 1 AS `{keyword}`");
            assert!(!is_write_sql(&sql), "expected read-only SQL: {sql}");
        }
        assert!(!is_write_sql("SHOW COLUMNS FROM `delete`"));
        assert!(!is_write_sql("DESC `delete`"));
        assert!(!is_write_sql("SELECT 1 AS `before``delete`"));
        assert!(!is_write_sql("SELECT 1 AS `semi;delete`; SELECT 2"));
    }

    #[test]
    fn is_write_sql_still_blocks_writes_with_backtick_identifiers() {
        assert!(is_write_sql("DELETE FROM `users`"));
        assert!(is_write_sql("DROP TABLE `users`"));
        assert!(is_write_sql("UPDATE `users` SET name = 'Ada'"));
    }

    #[test]
    fn is_write_sql_allows_show_create_statements() {
        for sql in [
            "SHOW CREATE TABLE users",
            "show create view active_users",
            "SHOW CREATE PROCEDURE refresh_users",
            "SHOW CREATE FUNCTION user_count",
            "  /* metadata */\nSHOW\nCREATE TABLE users;",
            "SHOW CREATE TABLE users; SELECT ';' AS separator",
            "SHOW CREATE TABLE users /* ; DELETE FROM users */",
        ] {
            assert!(!is_write_sql(sql), "expected read-only SQL: {sql}");
        }
    }

    #[test]
    fn is_write_sql_blocks_writes_after_show_create() {
        for write_sql in [
            "DROP TABLE users",
            "DELETE FROM users",
            "TRUNCATE TABLE users",
            "ALTER TABLE users ADD COLUMN active BOOLEAN",
            "UPDATE users SET active = true",
            "MERGE INTO users USING source ON users.id = source.id WHEN MATCHED THEN UPDATE SET active = true",
            "REPLACE INTO users VALUES (1)",
            "INSERT INTO users VALUES (1)",
            "CREATE TABLE audit (id INT)",
        ] {
            let sql = format!("SHOW CREATE TABLE users; {write_sql}");
            assert!(is_write_sql(&sql), "expected write SQL: {sql}");
        }
    }

    #[test]
    fn is_write_sql_ignores_leading_whitespace_and_comments() {
        assert!(!is_write_sql("   /* comment */ SELECT * FROM users"));
        assert!(!is_write_sql("-- comment\nSELECT * FROM users"));
        assert!(is_write_sql("   /* comment */ INSERT INTO users VALUES (1)"));
    }

    #[test]
    fn is_write_sql_blocks_mysql_and_mariadb_executable_comments() {
        for sql in [
            "SELECT 3156 /*! INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
            "SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
            "SELECT 3156 /*M! INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
            "SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
        ] {
            assert!(is_write_sql_for_database(sql, DatabaseType::Mysql), "expected write SQL: {sql}");
        }

        assert!(!is_write_sql_for_database(
            "SELECT 3156 /* INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
            DatabaseType::Mysql
        ));
        assert!(!is_write_sql_for_database(
            "SELECT '/*!50000 INTO OUTFILE \'/tmp/probe\' */' AS note",
            DatabaseType::Mysql
        ));
    }

    #[test]
    fn is_write_sql_treats_mysql_executable_comment_syntax_as_plain_for_other_dialects() {
        for database_type in [DatabaseType::Postgres, DatabaseType::Sqlite] {
            for sql in [
                "SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
                "SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
                "SELECT 3156 /* ordinary block comment */",
            ] {
                assert!(
                    !is_write_sql_for_database(sql, database_type),
                    "expected read-only SQL for {database_type:?}: {sql}"
                );
            }
        }
    }

    #[test]
    fn is_write_sql_blocks_dialect_specific_select_into_writes() {
        for sql in [
            "SELECT 3156 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'",
            "SELECT 3156 INTO DUMPFILE '/var/lib/mysql-files/dbx_ro_probe.bin'",
            "WITH probe AS (SELECT 3156) SELECT * FROM probe INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'",
        ] {
            assert!(is_write_sql_for_database(sql, DatabaseType::Mysql), "expected write SQL: {sql}");
        }

        for database_type in [
            DatabaseType::Postgres,
            DatabaseType::Redshift,
            DatabaseType::Gaussdb,
            DatabaseType::OpenGauss,
            DatabaseType::Kingbase,
            DatabaseType::Highgo,
            DatabaseType::Vastbase,
            DatabaseType::Kwdb,
        ] {
            for sql in [
                "SELECT * INTO copied_users FROM users",
                "WITH active AS (SELECT * FROM users WHERE active) SELECT * INTO active_users FROM active",
            ] {
                assert!(
                    is_write_sql_for_database(sql, database_type),
                    "expected write SQL for {database_type:?}: {sql}"
                );
            }
        }

        for sql in [
            "SELECT * INTO dbo.copied_users FROM dbo.users",
            "WITH active AS (SELECT * FROM users WHERE active = 1) SELECT * INTO #active_users FROM active",
        ] {
            assert!(is_write_sql_for_database(sql, DatabaseType::SqlServer), "expected write SQL: {sql}");
        }
    }

    #[test]
    fn is_write_sql_does_not_globally_block_into() {
        for sql in [
            "SELECT 3156 INTO @probe",
            "SELECT 'INTO OUTFILE /tmp/probe' AS note",
            "SELECT 3156 /* INTO OUTFILE '/tmp/probe' */",
            "SELECT 1 AS `into`, 2 AS `outfile`",
        ] {
            assert!(!is_write_sql_for_database(sql, DatabaseType::Mysql), "expected read-only SQL: {sql}");
        }

        for sql in
            ["SELECT 'INTO copied_users' AS note", "SELECT $$INTO copied_users$$ AS note", "SELECT 1 AS \"into\""]
        {
            assert!(!is_write_sql_for_database(sql, DatabaseType::Postgres), "expected read-only SQL: {sql}");
        }

        assert!(!is_write_sql_for_database("SELECT 1 AS [into]", DatabaseType::SqlServer));
        assert!(!is_write_sql_for_database("SELECT 1 INTO unsupported", DatabaseType::Sqlite));
    }

    #[test]
    fn is_write_sql_cte_with_nested_write() {
        // CTE starting with WITH but containing a write operation inside
        assert!(is_write_sql("WITH deleted AS (DELETE FROM users RETURNING id) SELECT * FROM deleted"));
        assert!(is_write_sql("WITH updated AS (UPDATE users SET name = 'x' RETURNING id) SELECT * FROM updated"));
        assert!(is_write_sql("WITH inserted AS (INSERT INTO users VALUES (1) RETURNING id) SELECT * FROM inserted"));
        // Pure read CTE should be allowed
        assert!(!is_write_sql("WITH cte AS (SELECT * FROM users) SELECT * FROM cte"));
    }

    #[test]
    fn is_write_sql_case_insensitive() {
        assert!(!is_write_sql("select * from users"));
        assert!(!is_write_sql("Select * From users"));
        assert!(is_write_sql("insert into users values (1)"));
        assert!(is_write_sql("Insert Into users Values (1)"));
        assert!(is_write_sql("update users set name = 'x'"));
    }

    #[test]
    fn is_write_sql_edge_cases() {
        assert!(!is_write_sql("")); // empty string -> not a write
        assert!(!is_write_sql("   ")); // whitespace only -> not a write
        assert!(is_write_sql("COMMIT")); // not a recognized read keyword
        assert!(is_write_sql("ROLLBACK")); // not a recognized read keyword
        assert!(is_write_sql("BEGIN")); // not a recognized read keyword
        assert!(is_write_sql("GRANT SELECT ON users TO admin"));
        assert!(is_write_sql("REVOKE SELECT ON users FROM admin"));
    }

    #[test]
    fn is_write_sql_blocks_stored_procedure_calls() {
        // CALL and EXEC don't start with a read keyword, so they are treated as writes
        assert!(is_write_sql("CALL my_procedure(1, 2)"));
        assert!(is_write_sql("call my_procedure()"));
        assert!(is_write_sql("EXEC sp_update_stats"));
        assert!(is_write_sql("EXECUTE sp_rename 'old', 'new'"));
        assert!(is_write_sql("execute my_func()"));
    }

    #[test]
    fn is_write_sql_allows_safe_read_pragmas() {
        // Read-only PRAGMA forms (function-call style with known safe names) are allowed
        assert!(!is_write_sql("PRAGMA table_info(users)"));
        assert!(!is_write_sql("PRAGMA table_xinfo(users)"));
        assert!(!is_write_sql("PRAGMA index_list(users)"));
        assert!(!is_write_sql("PRAGMA index_info(idx_name)"));
        assert!(!is_write_sql("PRAGMA foreign_key_list(users)"));
        assert!(!is_write_sql("PRAGMA database_list"));
        assert!(!is_write_sql("PRAGMA compile_options"));
        assert!(!is_write_sql("PRAGMA data_version"));
        assert!(!is_write_sql("pragma table_info(users)"));
    }

    #[test]
    fn is_write_sql_blocks_unsafe_pragmas() {
        // Assignment forms are always blocked
        assert!(is_write_sql("PRAGMA journal_mode = WAL"));
        assert!(is_write_sql("PRAGMA synchronous = OFF"));
        assert!(is_write_sql("PRAGMA foreign_keys = ON"));
        assert!(is_write_sql("PRAGMA cache_size = -2000"));
        assert!(is_write_sql("PRAGMA user_version = 123"));
        // Unknown PRAGMA names are blocked
        assert!(is_write_sql("PRAGMA writable_schema = ON"));
        assert!(is_write_sql("PRAGMA locking_mode = EXCLUSIVE"));
        assert!(is_write_sql("PRAGMA temp_store = MEMORY"));
        assert!(is_write_sql("PRAGMA some_unknown_pragma"));
    }

    #[test]
    fn is_write_sql_string_literal_hides_keywords() {
        // The dangerous keyword is inside a string literal, so it should NOT be detected
        assert!(!is_write_sql("SELECT 'DROP TABLE users' AS hint FROM t"));
        assert!(!is_write_sql("SELECT * FROM t WHERE name = 'delete'"));
    }

    #[test]
    fn check_read_only_success_and_error() {
        assert_eq!(check_read_only("SELECT * FROM users", "prod-db", DatabaseType::Mysql), Ok(()));
        assert_eq!(check_read_only("WITH cte AS (SELECT 1) SELECT * FROM cte", "prod-db", DatabaseType::Mysql), Ok(()));
        assert_eq!(check_read_only("SHOW CREATE TABLE users", "prod-db", DatabaseType::Mysql), Ok(()));
        assert_eq!(check_read_only("SELECT 1 AS `delete`", "prod-db", DatabaseType::Mysql), Ok(()));

        let err = check_read_only("DELETE FROM users", "prod-db", DatabaseType::Mysql);
        assert!(err.is_err());
        assert_eq!(
            err.unwrap_err(),
            "Read-only mode: connection 'prod-db' has read-only protection enabled. Write operation (including stored procedure calls) blocked."
        );

        let err2 = check_read_only("UPDATE users SET name = 'x'", "reporting-db", DatabaseType::Mysql);
        assert!(err2.is_err());
        assert!(err2.unwrap_err().contains("reporting-db"));

        let show_create_err =
            check_read_only("SHOW CREATE TABLE users; DELETE FROM users", "prod-db", DatabaseType::Mysql);
        assert!(show_create_err.is_err());
        assert!(show_create_err.unwrap_err().contains("Write operation"));
    }

    #[test]
    fn check_read_only_allows_mysql_checksum_only_for_mysql() {
        assert_eq!(check_read_only("CHECKSUM TABLE `issue6693_repro`", "mysql", DatabaseType::Mysql), Ok(()));
        assert!(check_read_only("CHECKSUM TABLE users", "postgres", DatabaseType::Postgres).is_err());
        assert!(check_read_only("CHECKSUM TABLE users", "sqlite", DatabaseType::Sqlite).is_err());
        assert!(check_read_only("CHECKSUM TABLE users; DROP TABLE users", "mysql", DatabaseType::Mysql).is_err());
    }

    #[test]
    fn check_read_only_allows_only_sqlserver_plan_capture_session_switches() {
        for sql in [
            "SET SHOWPLAN_XML ON;",
            "-- explain\nSET SHOWPLAN_XML OFF",
            "SET STATISTICS XML ON;",
            "-- actual plan\nSET STATISTICS XML OFF",
        ] {
            assert_eq!(check_read_only(sql, "readonly", DatabaseType::SqlServer), Ok(()));
        }
        assert!(check_read_only("SET SHOWPLAN_ALL ON", "readonly", DatabaseType::SqlServer).is_err());
        assert!(check_read_only("SET STATISTICS IO ON", "readonly", DatabaseType::SqlServer).is_err());
        assert!(check_read_only("SET SHOWPLAN_XML ON; SELECT 1", "readonly", DatabaseType::SqlServer).is_err());
        assert!(check_read_only("SET SHOWPLAN_XML OFF; DROP TABLE users", "readonly", DatabaseType::SqlServer).is_err());
        assert!(
            check_read_only("SET STATISTICS XML OFF; DROP TABLE users", "readonly", DatabaseType::SqlServer).is_err()
        );
    }

    #[test]
    fn check_read_only_only_treats_executable_comments_as_writes_for_mysql_compatible_connections() {
        let mysql_executable_comment = "SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */";
        let mariadb_executable_comment =
            "SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */";

        assert!(check_read_only(mysql_executable_comment, "mysql", DatabaseType::Mysql).is_err());
        assert!(check_read_only(mariadb_executable_comment, "mariadb", DatabaseType::Mysql).is_err());

        for database_type in [DatabaseType::Postgres, DatabaseType::Sqlite] {
            assert_eq!(check_read_only(mysql_executable_comment, "readonly", database_type), Ok(()));
            assert_eq!(check_read_only(mariadb_executable_comment, "readonly", database_type), Ok(()));
        }
    }

    #[test]
    fn check_read_only_blocks_dialect_specific_select_into_writes() {
        for (sql, database_type) in [
            ("SELECT 3156 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'", DatabaseType::Mysql),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Postgres),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Redshift),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Gaussdb),
            ("SELECT * INTO copied_users FROM users", DatabaseType::OpenGauss),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Kingbase),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Highgo),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Vastbase),
            ("SELECT * INTO copied_users FROM users", DatabaseType::Kwdb),
            ("SELECT * INTO #copied_users FROM users", DatabaseType::SqlServer),
        ] {
            assert!(check_read_only(sql, "readonly", database_type).is_err(), "expected blocked SQL: {sql}");
        }
    }

    #[test]
    fn strip_sql_comments_basic() {
        assert_eq!(strip_sql_comments("SELECT 1"), "SELECT 1");
        assert_eq!(strip_sql_comments("-- comment\nSELECT 1"), " SELECT 1");
        assert_eq!(strip_sql_comments("/* block */ SELECT 1"), "  SELECT 1");
        assert_eq!(strip_sql_comments("# comment\nSELECT 1"), " SELECT 1");
    }

    #[test]
    fn strip_sql_comments_preserves_strings() {
        // strip_sql_comments does NOT handle string delimiters, so it strips
        // comments even inside string literals
        assert_eq!(strip_sql_comments("SELECT 'hello /* not a comment */'"), "SELECT 'hello  '");
    }

    #[test]
    fn classifies_dynamodb_generated_statements_for_read_only_protection() {
        assert!(!is_write_sql_for_database(
            "DBX DYNAMODB SCAN\ntable: \"orders\"\nlimit: 1000",
            DatabaseType::DynamoDb
        ));
        assert!(!is_write_sql_for_database(
            "DBX DYNAMODB QUERY / SCAN\ntable: \"orders\"\nfilter:\n{\"status\":\"SHIPPED\"}",
            DatabaseType::DynamoDb
        ));
        for operation in ["INSERT ITEM", "PUT ITEM", "DELETE ITEM", "UNKNOWN"] {
            assert!(is_write_sql_for_database(
                &format!("DBX DYNAMODB {operation}\ntable: \"orders\""),
                DatabaseType::DynamoDb
            ));
        }
    }

    #[test]
    fn classifies_search_engine_rest_writes() {
        assert!(!is_write_sql_for_database("GET /_cluster/health", DatabaseType::Easysearch));
        assert!(!is_write_sql_for_database(
            "POST /products/_search\n{\"query\":{\"match_all\":{}}}",
            DatabaseType::Easysearch
        ));
        assert!(is_write_sql_for_database(
            "PUT /products/_doc/1?refresh=true\n{\"name\":\"Notebook\"}",
            DatabaseType::Easysearch
        ));
    }

    #[test]
    fn classifies_solr_rest_requests() {
        // Read-only requests: select/query/get and other read handlers.
        assert!(!is_write_sql_for_database("GET /mycore/select?q=*:*&rows=20", DatabaseType::Solr));
        assert!(!is_write_sql_for_database(
            "POST /mycore/query\n{\"query\":\"name:foo\",\"limit\":10}",
            DatabaseType::Solr
        ));
        assert!(!is_write_sql_for_database("GET /mycore/schema/fields", DatabaseType::Solr));

        // Document writes hit the update handlers, including the `/update/...`
        // variants whose last path segment is not literally `update`.
        assert!(is_write_sql_for_database(
            "POST /mycore/update\n{\"add\":{\"doc\":{\"id\":\"1\"}}}",
            DatabaseType::Solr
        ));
        assert!(is_write_sql_for_database("POST /mycore/update/json/docs\n{\"id\":\"1\"}", DatabaseType::Solr));
        assert!(is_write_sql_for_database("GET /mycore/update?commit=true", DatabaseType::Solr));

        // CoreAdmin actions mutate over GET too; only the read whitelist stays safe.
        assert_eq!(
            classify_search_engine_query_risk("GET /admin/cores?action=CREATE&name=x", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /admin/cores?action=STATUS", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /admin/info/system", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );

        // Replication commands mutate over GET too; only the read whitelist stays safe.
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=details", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=restorestatus", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=filelist", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=disablereplication", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=enablereplication", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk("GET /mycore/replication?command=fetchindex", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::Dangerous)
        );

        // Core/schema/admin operations are dangerous.
        assert_eq!(
            classify_search_engine_query_risk("POST /admin/cores?action=CREATE&name=x", DatabaseType::Solr),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk(
                "POST /mycore/schema\n{\"add-field\":{\"name\":\"x\",\"type\":\"string\"}}",
                DatabaseType::Solr
            ),
            Some(SearchEngineQueryRisk::Dangerous)
        );

        // Non-Solr types never take this classifier.
        assert!(classify_search_engine_query_risk("POST /mycore/update", DatabaseType::Postgres).is_none());
    }

    #[test]
    fn excludes_victoriametrics_from_sql_query_paths() {
        assert!(!supports_sql_query(DatabaseType::VictoriaMetrics));
        assert!(supports_sql_query(DatabaseType::Postgres));
        assert_eq!(
            check_read_only(
                r#"{__name__="dbx_issue_5352_temperature_celsius"}[5m]"#,
                "VictoriaMetrics",
                DatabaseType::VictoriaMetrics,
            ),
            Ok(())
        );
    }

    #[test]
    fn mysql_explicit_transaction_openers_are_recognized() {
        for sql in [
            "BEGIN",
            "begin",
            "BEGIN;",
            "  BEGIN  ",
            "BEGIN WORK",
            "BEGIN WORK;",
            "-- open a transaction\nBEGIN",
            "/* keep */ START TRANSACTION",
            "START TRANSACTION",
            "start transaction;",
            "START TRANSACTION READ ONLY",
            "START TRANSACTION READ WRITE",
            "START TRANSACTION WITH CONSISTENT SNAPSHOT",
        ] {
            assert!(mysql_statement_opens_explicit_transaction(sql), "expected opener: {sql}");
        }
    }

    #[test]
    fn mysql_non_openers_are_not_treated_as_explicit_transactions() {
        for sql in [
            "",
            "   ",
            ";",
            "SELECT 1",
            "COMMIT",
            "ROLLBACK",
            "BEGIN;\nUPDATE t SET a = 1;",
            "BEGIN\nDECLARE x INT;\nEND",
            "BEGIN END",
            "START REPLICA",
            "SET autocommit = 0",
            "SELECT 'BEGIN' FROM t",
        ] {
            assert!(!mysql_statement_opens_explicit_transaction(sql), "expected non-opener: {sql}");
        }
    }
}