dbx/crates/dbx-sql-core/src/query_execution_sql.rs

use serde::{Deserialize, Serialize};
use sqlparser::dialect::{GenericDialect, 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_strategy_matches_capability_sql_format_and_acquisition` 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",
        }
    }
}

/// The execution contract used to acquire an estimated plan.
///
/// This describes the acquisition path only; connection and driver execution
/// remain owned by `dbx-core`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EstimatedPlanAcquisition {
    /// Execute the SQL generated by `build_explain_sql`.
    GeneratedSql,
    /// Ask the connected driver's existing explain facility for a plan.
    DriverNative,
    /// Capture SHOWPLAN on a dedicated SQL Server client session.
    SqlServerShowPlanSession,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EstimatedPlanSqlGeneration {
    Postgres,
    Mysql,
    Plain,
    Xugu,
    Oracle,
    Db2,
    SqlServer,
    Json,
}

/// One complete Estimated Plan contract for a database dialect.
///
/// Capability, SQL generation, payload encoding and Plugin Host acquisition
/// all derive from this mapping.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EstimatedPlanStrategy {
    sql_generation: EstimatedPlanSqlGeneration,
    acquisition: EstimatedPlanAcquisition,
    format: EstimatedPlanFormat,
}

impl EstimatedPlanStrategy {
    pub const fn acquisition(self) -> EstimatedPlanAcquisition {
        self.acquisition
    }

    pub const fn format(self) -> EstimatedPlanFormat {
        self.format
    }
}

#[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 {
    let Some(strategy) = estimated_plan_strategy(options.database_type) else {
        return explain_err("unsupported");
    };
    build_explain_sql_with_strategy(options, strategy)
}

/// Builds the only validated GaussDB(DWS) Estimated Plan form. This does not
/// make `DatabaseType::Gaussdb` generally supported: callers must first verify
/// the explicit DWS connection profile and supported server version.
pub fn build_gaussdb_dws_explain_sql(options: ExplainSqlOptions) -> ExplainSqlBuildResult {
    if options.database_type != Some(DatabaseType::Gaussdb) {
        return explain_err("unsupported");
    }
    if options.analyze == Some(true) {
        return explain_err("unsafe");
    }
    build_explain_sql_with_strategy(options, gaussdb_dws_estimated_plan_strategy())
}

fn build_explain_sql_with_strategy(
    options: ExplainSqlOptions,
    strategy: EstimatedPlanStrategy,
) -> ExplainSqlBuildResult {
    let source = strip_trailing_semicolons(options.sql.trim());
    if source.is_empty() {
        return explain_err("empty");
    }
    let safety_source =
        if options.database_type == Some(DatabaseType::Db2) { options.sql.as_str() } else { source.as_str() };
    if !is_safe_explain_sql_for_database(safety_source, options.database_type) {
        return explain_err("unsafe");
    }
    if options.database_type == Some(DatabaseType::Xugu) {
        // Xugu's documented SELECT forms; do not inherit PG-only TABLE/VALUES
        // shortcuts or permit DML/Autotrace in the first visual-plan implementation.
        let leading = strip_sql_comments(&source).trim_start().to_lowercase();
        if !["select", "with"].iter().any(|keyword| starts_with_explain_keyword_boundary(&leading, keyword)) {
            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 strategy.sql_generation {
        // ANALYZE executes the statement; the safety gate above has already
        // limited the source to the existing read-only subset.
        EstimatedPlanSqlGeneration::Postgres if options.analyze == Some(true) => {
            format!("EXPLAIN (ANALYZE, FORMAT JSON) {source}")
        }
        EstimatedPlanSqlGeneration::Postgres => format!("EXPLAIN (FORMAT JSON) {source}"),
        EstimatedPlanSqlGeneration::Plain => format!("EXPLAIN {source}"),
        EstimatedPlanSqlGeneration::Xugu => format!("EXPLAIN VERBOSE {source}"),
        EstimatedPlanSqlGeneration::Oracle | EstimatedPlanSqlGeneration::Db2 => 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.
        EstimatedPlanSqlGeneration::SqlServer if options.analyze == Some(true) => {
            format!("SET STATISTICS XML ON;\nGO\n{source}\nGO\nSET STATISTICS XML OFF;")
        }
        EstimatedPlanSqlGeneration::SqlServer => {
            format!("SET SHOWPLAN_XML ON;\nGO\n{source}\nGO\nSET SHOWPLAN_XML OFF;")
        }
        EstimatedPlanSqlGeneration::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}")
        }
        EstimatedPlanSqlGeneration::Mysql | EstimatedPlanSqlGeneration::Json => {
            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
}

/// The estimated-plan contract for a database, if one is currently supported.
pub fn estimated_plan_strategy(database_type: Option<DatabaseType>) -> Option<EstimatedPlanStrategy> {
    use EstimatedPlanAcquisition::{DriverNative, GeneratedSql, SqlServerShowPlanSession};
    use EstimatedPlanFormat::{Json, Text, Xml};
    use EstimatedPlanSqlGeneration::{Db2, Mysql, Oracle, Plain, Postgres, SqlServer, Xugu};

    let strategy = match database_type? {
        DatabaseType::Db2 => EstimatedPlanStrategy { sql_generation: Db2, acquisition: DriverNative, format: Json },
        DatabaseType::Mysql => EstimatedPlanStrategy { sql_generation: Mysql, acquisition: GeneratedSql, format: Json },
        DatabaseType::Doris => EstimatedPlanStrategy { sql_generation: Plain, acquisition: GeneratedSql, format: Text },
        DatabaseType::Xugu => EstimatedPlanStrategy { sql_generation: Xugu, acquisition: GeneratedSql, format: Text },
        DatabaseType::Postgres => {
            EstimatedPlanStrategy { sql_generation: Postgres, acquisition: GeneratedSql, format: Json }
        }
        // openGauss documents EXPLAIN (FORMAT TEXT|XML|JSON|YAML) but JSON output
        // additionally depends on the session's explain_perf_mode; plain text avoids
        // assuming that GUC, so the plan is generated as plain `EXPLAIN <statement>`
        // text (#13055).
        DatabaseType::OpenGauss => {
            EstimatedPlanStrategy { sql_generation: Plain, acquisition: GeneratedSql, format: Text }
        }
        DatabaseType::Questdb => {
            EstimatedPlanStrategy { sql_generation: Plain, acquisition: GeneratedSql, format: Text }
        }
        DatabaseType::Dameng => {
            EstimatedPlanStrategy { sql_generation: Plain, acquisition: DriverNative, format: Text }
        }
        DatabaseType::Oracle => {
            EstimatedPlanStrategy { sql_generation: Oracle, acquisition: DriverNative, format: Text }
        }
        DatabaseType::OceanbaseOracle => EstimatedPlanStrategy {
            sql_generation: EstimatedPlanSqlGeneration::Json,
            acquisition: GeneratedSql,
            format: Json,
        },
        DatabaseType::SqlServer => {
            EstimatedPlanStrategy { sql_generation: SqlServer, acquisition: SqlServerShowPlanSession, format: Xml }
        }
        // The former MongoDB SQL-generation arm was unreachable behind the old
        // capability gate; it remains unsupported and has no Estimated Plan strategy.
        _ => return None,
    };
    Some(strategy)
}

/// The version-gated strategy used only for an explicitly identified and
/// validated Huawei DWS connection. Generic GaussDB remains unsupported.
pub const fn gaussdb_dws_estimated_plan_strategy() -> EstimatedPlanStrategy {
    EstimatedPlanStrategy {
        sql_generation: EstimatedPlanSqlGeneration::Postgres,
        acquisition: EstimatedPlanAcquisition::GeneratedSql,
        format: EstimatedPlanFormat::Json,
    }
}

/// Payload encoding defined by the estimated-plan strategy for this dialect.
/// Keep the historical JSON fallback for unsupported/unspecified dialects;
/// callers must check capability separately before using it.
pub fn estimated_plan_format(database_type: Option<DatabaseType>) -> EstimatedPlanFormat {
    estimated_plan_strategy(database_type).map_or(EstimatedPlanFormat::Json, EstimatedPlanStrategy::format)
}

pub fn supports_explain_plan(database_type: Option<DatabaseType>) -> bool {
    estimated_plan_strategy(database_type).is_some()
}

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 {
    if database_type == Some(DatabaseType::Db2) {
        return is_safe_db2_explain_sql(sql);
    }
    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)
}

/// Use the existing lexer so nested DB2 comments are removed before every
/// safety check; preserve literal escaping while normalizing only for validation.
fn is_safe_db2_explain_sql(sql: &str) -> bool {
    let Ok(tokens) = Tokenizer::new(&GenericDialect {}, sql).with_unescape(false).tokenize() else {
        return false;
    };
    let mut tokens: Vec<_> = tokens.into_iter().filter(|token| !matches!(token, Token::Whitespace(_))).collect();
    if tokens.last() == Some(&Token::SemiColon) {
        tokens.pop();
    }
    if tokens.iter().any(|token| matches!(token, Token::SemiColon)) {
        return false;
    }
    let Some(Token::Word(word)) = tokens.first() else {
        return false;
    };
    if word.quote_style.is_some() || !["SELECT", "WITH", "VALUES"].contains(&word.value.to_ascii_uppercase().as_str()) {
        return false;
    }
    let normalized = tokens.iter().map(ToString::to_string).collect::<Vec<_>>().join(" ");
    !contains_dangerous_sql_keyword(&normalized)
}

/// 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::CouchDb
            | DatabaseType::Qdrant
            | DatabaseType::Milvus
            | DatabaseType::Weaviate
            | DatabaseType::ChromaDb
            | DatabaseType::InfluxDb
            | DatabaseType::InfluxDb3
            | DatabaseType::VictoriaMetrics
            | DatabaseType::Neo4j
            | DatabaseType::Nebula
            | 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 {
    if database_type == DatabaseType::Neo4j {
        return !crate::cypher_read_only::is_proven_read_only_cypher(sql);
    }
    // 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 database_type == DatabaseType::Salesforce {
        // SOQL has no write verbs — every write arrives as a DBX pseudo-command —
        // so the keyword scan below would misread a string literal
        // (`Subject = 'Delete request'`) or a custom field (`Update__c`) as a
        // write and lock a read-only session out of an ordinary SELECT.
        return classify_salesforce_statement_risk(sql) != SalesforceStatementRisk::Read;
    }
    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,
    }
}

/// How a Salesforce statement must be treated by write and risk classification.
///
/// `SfClient::execute_query` accepts exactly two shapes: a SOQL `SELECT`, which
/// the REST query endpoint runs read-only, and a `DBX SALESFORCE <…>`
/// pseudo-command carrying a JSON body. Anything else is rejected by Salesforce
/// anyway, so it is classified as an opaque write and fails closed rather than
/// being guessed at by the SQL keyword scan.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SalesforceStatementRisk {
    /// SOQL `SELECT` — always a read, whatever its literals and field names say.
    Read,
    /// Single-record DML naming its object and, for update/delete, its record Id.
    ScopedWrite,
    /// Unscoped or unknown DML, unrecognized pseudo-commands, non-SOQL text.
    OpaqueWrite,
}

pub fn classify_salesforce_statement_risk(source: &str) -> SalesforceStatementRisk {
    use SalesforceStatementRisk::{OpaqueWrite, Read, ScopedWrite};

    let header = source.lines().find(|line| !line.trim().is_empty()).unwrap_or("").trim();
    if header.to_ascii_uppercase().starts_with("DBX SALESFORCE") {
        return if salesforce_dml_is_scoped(source) { ScopedWrite } else { OpaqueWrite };
    }
    // Comments are stripped before the verb check, but the pseudo-command branch
    // above deliberately reads the raw text: a `--` inside a JSON field value
    // must not be mistaken for a line comment and truncate the payload.
    let first_word = strip_sql_comments(source)
        .split(|character: char| !character.is_alphanumeric() && character != '_')
        .find(|token| !token.is_empty())
        .unwrap_or("")
        .to_ascii_uppercase();
    if first_word == "SELECT" {
        Read
    } else {
        OpaqueWrite
    }
}

/// Whether a `DBX SALESFORCE DML` body targets exactly one identified record:
/// an insert needs only its object, while an update or delete without an Id has
/// nothing bounding it and stays opaque.
fn salesforce_dml_is_scoped(source: &str) -> bool {
    // The header line never contains a brace, so the JSON body starts at the first `{`.
    let Some(start) = source.find('{') else { return false };
    let Ok(body) = serde_json::from_str::<serde_json::Value>(&source[start..]) else {
        return false;
    };
    let names_object =
        body.get("object").and_then(|value| value.as_str()).is_some_and(|object| !object.trim().is_empty());
    if !names_object {
        return false;
    }
    match body.get("op").and_then(|value| value.as_str()).map(str::to_ascii_lowercase).as_deref() {
        Some("insert") => true,
        Some("update") | Some("delete") => {
            body.get("id").and_then(|value| value.as_str()).is_some_and(|id| !id.trim().is_empty())
        }
        _ => false,
    }
}

#[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::CouchDb {
        return classify_couchdb_query_risk(source);
    }
    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,
    }
}

fn classify_couchdb_query_risk(source: &str) -> Option<SearchEngineQueryRisk> {
    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 segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();

    match method.as_str() {
        "GET" | "HEAD" | "OPTIONS" => Some(SearchEngineQueryRisk::ReadOnly),
        "POST"
            if segments.last().is_some_and(|s| {
                s.eq_ignore_ascii_case("_find")
                    || s.eq_ignore_ascii_case("_explain")
                    || s.eq_ignore_ascii_case("_all_docs")
                    || s.eq_ignore_ascii_case("_bulk_get")
            }) =>
        {
            Some(SearchEngineQueryRisk::ReadOnly)
        }
        // 视图查询（POST /{db}/_design/{doc}/_view/{name} 带 keys 等参数）只读。
        "POST"
            if segments
                .windows(3)
                .any(|w| w[0].eq_ignore_ascii_case("_design") && w[2].eq_ignore_ascii_case("_view")) =>
        {
            Some(SearchEngineQueryRisk::ReadOnly)
        }
        "DELETE" if segments.len() <= 1 => Some(SearchEngineQueryRisk::Dangerous),
        "DELETE" => Some(SearchEngineQueryRisk::Write),
        "PUT" if segments.len() <= 1 => Some(SearchEngineQueryRisk::Dangerous),
        "PUT" | "POST" => Some(SearchEngineQueryRisk::Write),
        _ => 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::Gbase8c
            | 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, 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, true);
    // 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
}

/// True when `upper` starts with `keyword` followed by a non-alphanumeric byte
/// (or nothing). `upper` must already be ASCII-uppercased. Shared with the
/// Oracle manual-transaction proof in [`crate::sql_risk`].
pub(crate) 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: true only for
/// a top-level read DBX can prove harmless.
///
/// This is the public Oracle entry point; the rule set lives in
/// [`crate::sql_risk::prove_read_only_for_database`] (see
/// `prove_oracle_read_only_statement` there) and combines the Oracle lexical
/// guard rails — no `WITH`, no row locking, no transaction control, no DDL/DCL,
/// no `CALL`/`EXEC`, no PL/SQL block, no `EXECUTE IMMEDIATE`, no database link
/// (`@`), no sequence pseudocolumn (`NEXTVAL`/`CURRVAL`), no `SELECT ... INTO` —
/// with an `OracleDialect` AST proof that every function call is on the reviewed
/// built-in whitelist.
///
/// Before #11788 the guard rails included "reject any `(`", which rejected every
/// ordinary query that used a built-in function (`SELECT LENGTH(1) FROM dual`)
/// and could not tell a call from a datatype argument
/// (`CAST(a AS VARCHAR2(10))`) or an `IN (1, 2)` list.
pub fn is_oracle_proven_read_only_statement(sql: &str) -> bool {
    crate::sql_risk::prove_read_only_for_database(sql, DatabaseType::Oracle)
        == crate::sql_risk::ReadProof::ProvenReadOnly
}

/// Match a whole word (`word` bounded by non-identifier bytes) in a
/// pre-lowercased, comment/literal-stripped source. Shared with the Oracle
/// manual-transaction proof in [`crate::sql_risk`].
pub(crate) 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, true).0
}

/// [`strip_sql_comments_and_literals`] with `#` handled as a **code character**
/// instead of a line-comment start.
///
/// `#` only starts a comment in the MySQL family; Oracle (and OceanBase-Oracle)
/// allow `#` inside unquoted identifiers, so stripping from `#` to end of line
/// hides real code from the safety scanners. `sql::SqlDialectProfile::
/// supports_hash_line_comments` models the same distinction for statement
/// splitting. Only the Oracle manual-transaction proof opts in here; every other
/// caller keeps the MySQL-conservative default, so no other dialect's
/// classification changes.
pub(crate) fn strip_sql_comments_and_literals_keeping_hash(sql: &str) -> String {
    strip_sql_comments_and_literals_with_metadata(sql, false, false).0
}

fn strip_sql_comments_and_literals_with_metadata(
    sql: &str,
    detect_mysql_executable_comments: bool,
    hash_line_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;
    // The closing `$$` / `$tag$` of the dollar-quoted literal being skipped.
    let mut in_dollar_quote: Option<String> = None;
    let mut has_mysql_executable_comment = false;
    // Last character emitted from code position, used to keep a `$` that is
    // still part of an identifier (`foo$tag$bar`) from opening a literal.
    let mut previous_code_char: Option<char> = None;

    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 let Some(delimiter) = &in_dollar_quote {
            if ch == '$' {
                let mut probe = chars.clone();
                let mut matched = true;
                for expected in delimiter.chars().skip(1) {
                    match probe.next() {
                        Some(actual) if actual == expected => {}
                        _ => {
                            matched = false;
                            break;
                        }
                    }
                }
                if matched {
                    for _ in delimiter.chars().skip(1) {
                        chars.next();
                    }
                    in_dollar_quote = None;
                }
            }
            output.push(' ');
            continue;
        }

        if ch == '-' && chars.peek() == Some(&'-') {
            chars.next();
            in_line_comment = true;
            continue;
        }
        // `#` starts a line comment only in the MySQL family; elsewhere it is a
        // code character (Oracle allows it inside identifiers), so callers that
        // pass `hash_line_comments = false` must keep the text visible.
        if hash_line_comments && 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 matches!(ch, 'q' | 'Q' | 'n' | 'N') && !previous_code_char.is_some_and(is_oracle_identifier_char) {
            if let Some(length) = oracle_alternative_quote_len(ch, &chars) {
                for _ in 0..length {
                    chars.next();
                }
                output.push_str(&" ".repeat(length));
                previous_code_char = Some(' ');
                continue;
            }
        }
        if ch == '\'' {
            in_single_quote = true;
            output.push(' ');
            previous_code_char = Some(' ');
            continue;
        }
        if ch == '"' {
            in_double_quote = true;
            output.push(' ');
            previous_code_char = Some(' ');
            continue;
        }
        if ch == '`' {
            in_backtick_quote = true;
            output.push(' ');
            previous_code_char = Some(' ');
            continue;
        }
        if ch == '$' && !previous_code_char.is_some_and(is_identifier_char) {
            if let Some(delimiter) = dollar_quote_delimiter(&chars) {
                for _ in delimiter.chars().skip(1) {
                    chars.next();
                }
                in_dollar_quote = Some(delimiter);
                output.push(' ');
                previous_code_char = Some(' ');
                continue;
            }
        }

        output.push(ch);
        previous_code_char = Some(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 is_identifier_char(character: char) -> bool {
    character.is_ascii_alphanumeric() || character == '_' || character == '$'
}

/// Oracle identifiers may also contain `#` (and `$` above), so an introducer
/// directly after one of those characters does not start its own token.
fn is_oracle_identifier_char(character: char) -> bool {
    character.is_alphanumeric() || character == '_' || character == '$' || character == '#'
}

/// Length of an Oracle alternative-quoted literal — `q'#...#'`, `q'[...]'`,
/// `nq'{...}'` — where the delimiter may contain the quote character
/// unchanged. The literal ends at the first `closer + '`; Oracle has no escape
/// or nesting inside these literals. `ch` is the already-read `q`/`Q`/`n`/`N`
/// introducer and the returned length includes it. An unterminated literal
/// returns `None` so the text stays visible to the keyword scan and malformed
/// input still fails closed, matching the dollar-quote policy here. Mirrors
/// `readOracleAlternativeQuoteEnd` in `apps/desktop/src/lib/sql/sqlRisk.ts`.
fn oracle_alternative_quote_len(ch: char, chars: &std::iter::Peekable<std::str::Chars<'_>>) -> Option<usize> {
    let mut probe = chars.clone();
    let mut length = 1usize;
    if matches!(ch, 'n' | 'N') {
        match probe.next() {
            Some('q' | 'Q') => length += 1,
            _ => return None,
        }
    }
    match probe.next() {
        Some('\'') => length += 1,
        _ => return None,
    }
    let opener = probe.next()?;
    length += 1;
    // A whitespace opener means `q '...'` — an identifier then an ordinary
    // string, not an alternative-quoted literal.
    if opener.is_whitespace() {
        return None;
    }
    let closer = match opener {
        '[' => ']',
        '{' => '}',
        '(' => ')',
        '<' => '>',
        other => other,
    };
    loop {
        let scan = probe.next()?;
        length += 1;
        if scan == closer && probe.peek() == Some(&'\'') {
            probe.next();
            return Some(length + 1);
        }
    }
}

/// Read a PostgreSQL dollar-quote delimiter — `$$` or `$tag$` — from the
/// characters that follow an opening `$`, and return it only when the literal is
/// actually closed later in the statement.
///
/// The tag rules mirror PostgreSQL's own lexer (empty, or an identifier that
/// does not start with a digit) and sqlparser's dollar-quote tokenizer, so the
/// keyword scan agrees with the AST classifier in `sql_risk` instead of reading
/// the body of a literal as SQL. Requiring a matching closing delimiter keeps an
/// unterminated `$$` — which the database rejects anyway — visible to the
/// keyword scan, so malformed input still fails closed.
fn dollar_quote_delimiter(chars: &std::iter::Peekable<std::str::Chars<'_>>) -> Option<String> {
    let mut probe = chars.clone();
    let delimiter = if probe.peek() == Some(&'$') {
        "$$".to_string()
    } else {
        let mut tag = String::new();
        while probe.peek().is_some_and(|ch| ch.is_ascii_alphanumeric() || *ch == '_') {
            tag.push(*probe.peek()?);
            probe.next();
        }
        if tag.is_empty() || tag.starts_with(|ch: char| ch.is_ascii_digit()) || probe.peek() != Some(&'$') {
            return None;
        }
        format!("${tag}$")
    };
    for _ in delimiter.chars().skip(1) {
        probe.next();
    }
    dollar_quote_is_closed(&mut probe, &delimiter).then_some(delimiter)
}

/// True when `delimiter` occurs again in `chars`, which is positioned just after
/// an opening delimiter.
fn dollar_quote_is_closed(chars: &mut std::iter::Peekable<std::str::Chars<'_>>, delimiter: &str) -> bool {
    let delimiter: Vec<char> = delimiter.chars().collect();
    let mut window: Vec<char> = Vec::with_capacity(delimiter.len());
    for character in chars.by_ref() {
        window.push(character);
        if window.len() > delimiter.len() {
            window.remove(0);
        }
        if window == delimiter {
            return true;
        }
    }
    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() {
        for sql in [
            "SELECT * FROM users",
            "select id, name from users",
            "SELECT * FROM users WHERE id = 1",
            "SELECT * FROM users ORDER BY id",
            "SELECT * FROM users FETCH FIRST 10 ROWS ONLY",
            "SELECT * FROM users ORDER BY id OFFSET 5 ROWS FETCH NEXT 10 ROWS ONLY",
            "SELECT 1",
            "SELECT ROWNUM FROM dual",
            "SELECT * FROM users;",
            // Comment/literal stripping hides keywords that are not part of the SQL.
            "-- comment\nSELECT * FROM users",
            "SELECT 'update' FROM dual",
            "SELECT \"select\" FROM t",
            "SELECT * FROM user_update",
        ] {
            assert!(is_oracle_proven_read_only_statement(sql), "expected proven: {sql}");
        }
    }

    #[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}");
        }
    }

    /// #11788: the old blanket "reject any `(`" rule rejected every ordinary
    /// read that used a built-in function. Whitelisted calls, non-call
    /// parentheses (`IN` lists, grouping, datatype arguments) and subqueries are
    /// proven now; the AST walk is what tells a call from a datatype argument.
    #[test]
    fn oracle_proven_read_only_accepts_whitelisted_functions_and_parenthesised_reads() {
        for sql in [
            "SELECT LENGTH(1) FROM dual",
            "SELECT COUNT(*) FROM users",
            "SELECT UPPER(name) FROM users",
            "SELECT NVL(name, 'x') FROM users",
            "SELECT TO_CHAR(SYSDATE, 'YYYY-MM-DD') FROM dual",
            "SELECT SUBSTR(name, 1, 3) || DECODE(a, 1, 'x', 'y') FROM users",
            "SELECT MIN(a), MAX(a), AVG(a), SUM(a) FROM t",
            "SELECT COUNT(*) FROM t GROUP BY a HAVING COUNT(*) > 1",
            "SELECT * FROM t WHERE id IN (1, 2, 3)",
            "SELECT (a + b) * c FROM t",
            "SELECT * FROM t WHERE (a = 1 OR b = 2)",
            "SELECT CAST(a AS VARCHAR2(10)) FROM t",
            "SELECT CAST(a AS NUMBER(10,2)) FROM t",
            "SELECT * FROM (SELECT * FROM users)",
            "SELECT * FROM users WHERE id IN (SELECT id FROM orders)",
            // Read-only keyword forms sqlparser models as function calls must
            // stay proven, or the fix would regress them.
            "SELECT CURRENT_DATE FROM dual",
            "SELECT CURRENT_TIME FROM dual",
            "SELECT CURRENT_TIMESTAMP FROM dual",
            "SELECT SYSTIMESTAMP, USER, LEVEL FROM dual",
            "SELECT LEVEL, a FROM t START WITH id = 1 CONNECT BY PRIOR id = pid",
            "SELECT /*+ FULL(t) */ COUNT(*) FROM t WHERE a = :1",
        ] {
            assert!(is_oracle_proven_read_only_statement(sql), "expected proven: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_unknown_functions_and_qualified_calls() {
        for sql in [
            // Not on the reviewed built-in whitelist.
            "SELECT my_udf(a) FROM t",
            "SELECT my_pkg.fn(a) FROM t",
            "SELECT DBMS_LOCK.SLEEP(1) FROM dual",
            "SELECT SYS_CONTEXT('USERENV', 'SESSION_USER') FROM dual",
            "SELECT SYS_GUID() FROM dual",
            "SELECT XMLAGG(x) FROM t",
            "SELECT LISTAGG(a, ',') WITHIN GROUP (ORDER BY a) FROM t",
            // Window and table functions stay Unproven by the visitor.
            "SELECT ROW_NUMBER() OVER (ORDER BY a) FROM t",
            "SELECT * FROM TABLE(my_func())",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    #[test]
    fn oracle_proven_read_only_rejects_empty_and_multi_statement_input() {
        for sql in [
            "",
            "   ",
            "-- only a comment",
            "SELECT 1 UNION SELECT 2",
            "SELECT 1; SELECT 2",
            "SELECT a FROM t; SELECT b FROM u",
            // Parse failure is judged by the lexical guard rails alone. Call-shaped
            // syntax stays Unproven because no AST can vouch for the call.
            "SELECT * FROM t AS OF TIMESTAMP TO_TIMESTAMP('2020-01-01', 'YYYY-MM-DD')",
        ] {
            assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
        }
    }

    /// Oracle-only syntax sqlparser cannot parse keeps the verdict it had before
    /// #11788 (no `(`, no `@`, no forbidden word) instead of losing the bit: the
    /// pre-change classifier never parsed the statement either.
    #[test]
    fn oracle_proven_read_only_keeps_pre_change_verdict_for_unparseable_oracle_syntax() {
        for sql in [
            "SELECT * FROM t AS OF TIMESTAMP SYSTIMESTAMP - INTERVAL '1' HOUR",
            "SELECT * FROM t AS OF SCN 12345",
            "SELECT * FROM t VERSIONS BETWEEN TIMESTAMP SYSTIMESTAMP - INTERVAL '1' HOUR AND SYSTIMESTAMP",
        ] {
            assert!(is_oracle_proven_read_only_statement(sql), "expected proven: {sql}");
        }
    }

    #[test]
    fn db2_explain_uses_native_json_plan_acquisition() {
        let db_type = Some(DatabaseType::Db2);
        assert!(supports_explain_plan(db_type));
        let strategy = estimated_plan_strategy(db_type).unwrap();
        assert_eq!(strategy.acquisition(), EstimatedPlanAcquisition::DriverNative);
        assert_eq!(strategy.format(), EstimatedPlanFormat::Json);
        for sql in [
            "SELECT * FROM \"APP\".\"ORDERS\"",
            "WITH q AS (SELECT 1 FROM SYSIBM.SYSDUMMY1) SELECT * FROM q",
            "VALUES 1",
            "/* plan */ select * FROM APP.ORDERS",
            "-- plan\nwith q AS (SELECT 1 FROM SYSIBM.SYSDUMMY1) SELECT * FROM q",
            "/* plan */ values (1)",
        ] {
            let result = build_explain_sql(ExplainSqlOptions {
                database_type: db_type,
                format: None,
                analyze: Some(true),
                sql: format!("{sql};"),
            });
            assert!(result.ok, "{result:?}");
            assert_eq!(result.sql, Some(format!("EXPLAIN PLAN FOR {sql}")));
        }
    }

    #[test]
    fn db2_explain_handles_nested_comments_without_hiding_writes() {
        for sql in [
            "/* outer /* inner */ trailing */ SELECT 1 FROM SYSIBM.SYSDUMMY1",
            "/* outer /* inner */ DELETE */ VALUES ('a;''b')",
            "WITH q AS (/* outer /* inner */ UPDATE */ SELECT 1 FROM SYSIBM.SYSDUMMY1) SELECT * FROM q",
        ] {
            assert!(is_safe_explain_sql_for_database(sql, Some(DatabaseType::Db2)), "{sql}");
            assert!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Db2),
                    format: None,
                    analyze: None,
                    sql: sql.into()
                })
                .ok,
                "{sql}"
            );
        }
        for sql in [
            "/* outer /* inner */ trailing */ DELETE FROM APP.ORDERS",
            "/* outer /* inner */ trailing */ SELECT 1; DELETE FROM APP.ORDERS",
            "/* outer /* inner */ trailing */ TABLE APP.ORDERS",
            "WITH q AS (DELETE FROM APP.ORDERS) SELECT * FROM q",
            "SELECT 1 /* outer /* inner */",
            "SELECT 1;;",
        ] {
            assert!(!is_safe_explain_sql_for_database(sql, Some(DatabaseType::Db2)), "{sql}");
            assert!(
                !build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Db2),
                    format: None,
                    analyze: None,
                    sql: sql.into()
                })
                .ok,
                "{sql}"
            );
        }
    }

    #[test]
    fn db2_explain_rejects_table_shortcuts_in_all_preflight_paths() {
        for sql in ["TABLE APP.ORDERS", "table \"APP\".\"ORDERS\";", "/* plan */ TABLE APP.ORDERS"] {
            assert!(!is_safe_explain_sql_for_database(sql, Some(DatabaseType::Db2)), "{sql}");
            let result = build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Db2),
                format: None,
                analyze: None,
                sql: sql.into(),
            });
            assert!(!result.ok, "{sql}: {result:?}");
            assert_eq!(result.reason.as_deref(), Some("unsafe"));
            assert!(is_safe_explain_sql(sql), "generic TABLE support must remain: {sql}");
            assert!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Postgres),
                    format: None,
                    analyze: None,
                    sql: sql.into(),
                })
                .ok,
                "PostgreSQL TABLE support must remain: {sql}"
            );
        }
    }

    #[test]
    fn db2_explain_rejects_empty_write_and_multiple_statement_sources() {
        for (sql, reason) in [
            ("", "empty"),
            ("DELETE FROM t", "unsafe"),
            ("UPDATE t SET n=1", "unsafe"),
            ("SELECT 1; DROP TABLE t", "unsafe"),
            ("SELECT 1; SELECT 2", "unsafe"),
        ] {
            let result = build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Db2),
                format: None,
                analyze: None,
                sql: sql.into(),
            });
            assert!(!result.ok);
            assert_eq!(result.reason.as_deref(), Some(reason));
        }
    }

    #[test]
    fn builds_xugu_verbose_text_explain_without_executing_the_query() {
        for sql in [
            "SELECT * FROM \"APP_TEST\".\"ORDERS\" WHERE id = 1",
            "WITH q AS (SELECT 1 FROM DUAL) SELECT * FROM q",
            "/* plan */ SELECT 1 FROM DUAL",
            "SELECT 'DELETE; DROP TABLE x' FROM DUAL",
        ] {
            for analyze in [None, Some(false), Some(true)] {
                let result = build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Xugu),
                    format: Some(ExplainFormat::Json),
                    analyze,
                    sql: format!(" {sql}; "),
                });
                assert_eq!(result.sql.as_deref(), Some(format!("EXPLAIN VERBOSE {sql}").as_str()));
                assert!(result.ok);
            }
        }
        assert_eq!(estimated_plan_format(Some(DatabaseType::Xugu)), EstimatedPlanFormat::Text);
        assert_eq!(
            estimated_plan_strategy(Some(DatabaseType::Xugu)).unwrap().acquisition(),
            EstimatedPlanAcquisition::GeneratedSql
        );
    }

    #[test]
    fn xugu_explain_rejects_writes_multiple_statements_and_unsupported_forms() {
        for sql in [
            "DELETE FROM t",
            "UPDATE t SET a = 1",
            "INSERT INTO t VALUES (1)",
            "MERGE INTO t USING s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.a = 1",
            "SELECT 1; DROP TABLE t",
            "WITH q AS (DELETE FROM t RETURNING *) SELECT * FROM q",
            "WITH FUNCTION f RETURN INT IS BEGIN RETURN 1; END; SELECT f() FROM DUAL",
            "EXPLAIN SELECT 1",
            "TABLE t",
            "VALUES (1)",
            "SELECTED",
            "",
        ] {
            let result = build_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Xugu),
                format: None,
                analyze: None,
                sql: sql.to_string(),
            });
            assert!(!result.ok, "unsafe Xugu explain source: {sql}");
            assert!(result.sql.is_none());
        }
    }

    #[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::Doris, "EXPLAIN SELECT 1"),
            (DatabaseType::Questdb, "EXPLAIN SELECT 1"),
            (DatabaseType::Dameng, "EXPLAIN SELECT 1"),
            (DatabaseType::Oracle, "EXPLAIN PLAN FOR SELECT 1"),
            (DatabaseType::OceanbaseOracle, "EXPLAIN FORMAT=JSON 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 by the strategy capability gate; its historical
            // SQL-generation arm remains unreachable.
            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 gaussdb_dws_strategy_is_explicit_and_keeps_the_generic_dialect_closed() {
        use EstimatedPlanAcquisition::GeneratedSql;
        use EstimatedPlanFormat::Json;

        let strategy = gaussdb_dws_estimated_plan_strategy();
        assert_eq!(strategy.acquisition(), GeneratedSql);
        assert_eq!(strategy.format(), Json);
        assert!(estimated_plan_strategy(Some(DatabaseType::Gaussdb)).is_none());
        assert!(!supports_explain_plan(Some(DatabaseType::Gaussdb)));

        let result = build_gaussdb_dws_explain_sql(ExplainSqlOptions {
            database_type: Some(DatabaseType::Gaussdb),
            format: None,
            analyze: None,
            sql: "SELECT 1".to_string(),
        });
        assert_eq!(result.sql.as_deref(), Some("EXPLAIN (FORMAT JSON) SELECT 1"));
        assert!(result.ok);

        for options in [
            ExplainSqlOptions {
                database_type: Some(DatabaseType::Gaussdb),
                format: None,
                analyze: Some(true),
                sql: "SELECT 1".to_string(),
            },
            ExplainSqlOptions {
                database_type: Some(DatabaseType::Gaussdb),
                format: None,
                analyze: None,
                sql: "DELETE FROM users".to_string(),
            },
        ] {
            assert!(!build_gaussdb_dws_explain_sql(options).ok);
        }

        assert_eq!(
            build_gaussdb_dws_explain_sql(ExplainSqlOptions {
                database_type: Some(DatabaseType::Postgres),
                format: None,
                analyze: None,
                sql: "SELECT 1".to_string(),
            })
            .reason
            .as_deref(),
            Some("unsupported")
        );
    }

    #[test]
    fn estimated_plan_strategy_matches_capability_sql_format_and_acquisition() {
        use EstimatedPlanAcquisition::{DriverNative, GeneratedSql, SqlServerShowPlanSession};
        use EstimatedPlanFormat::{Json, Text, Xml};

        let cases = [
            (DatabaseType::Mysql, "EXPLAIN FORMAT=JSON SELECT 1", Json, GeneratedSql),
            (DatabaseType::Doris, "EXPLAIN SELECT 1", Text, GeneratedSql),
            (DatabaseType::Xugu, "EXPLAIN VERBOSE SELECT 1", Text, GeneratedSql),
            (DatabaseType::Postgres, "EXPLAIN (FORMAT JSON) SELECT 1", Json, GeneratedSql),
            (DatabaseType::OpenGauss, "EXPLAIN SELECT 1", Text, GeneratedSql),
            (DatabaseType::Questdb, "EXPLAIN SELECT 1", Text, GeneratedSql),
            (DatabaseType::Dameng, "EXPLAIN SELECT 1", Text, DriverNative),
            (DatabaseType::Oracle, "EXPLAIN PLAN FOR SELECT 1", Text, DriverNative),
            (DatabaseType::Db2, "EXPLAIN PLAN FOR SELECT 1", Json, DriverNative),
            (DatabaseType::OceanbaseOracle, "EXPLAIN FORMAT=JSON SELECT 1", Json, GeneratedSql),
            (
                DatabaseType::SqlServer,
                "SET SHOWPLAN_XML ON;\nGO\nSELECT 1\nGO\nSET SHOWPLAN_XML OFF;",
                Xml,
                SqlServerShowPlanSession,
            ),
        ];

        for (database_type, expected_sql, expected_format, expected_acquisition) in cases {
            let strategy = estimated_plan_strategy(Some(database_type))
                .unwrap_or_else(|| panic!("{database_type:?} must have a complete Estimated Plan strategy"));
            assert!(supports_explain_plan(Some(database_type)), "{database_type:?}");
            assert_eq!(strategy.acquisition(), expected_acquisition, "{database_type:?}");
            assert_eq!(strategy.format(), expected_format, "{database_type:?}");
            assert_eq!(estimated_plan_format(Some(database_type)), expected_format, "{database_type:?}");

            let result = build_explain_sql(ExplainSqlOptions {
                database_type: Some(database_type),
                format: None,
                analyze: None,
                sql: "SELECT 1".to_string(),
            });
            assert!(result.ok, "{database_type:?}: {:?}", result.reason);
            assert_eq!(result.sql.as_deref(), Some(expected_sql), "{database_type:?}");
        }

        for database_type in [
            None,
            Some(DatabaseType::Redis),
            Some(DatabaseType::MongoDb),
            Some(DatabaseType::Gaussdb),
            Some(DatabaseType::StarRocks),
        ] {
            assert!(estimated_plan_strategy(database_type).is_none(), "{database_type:?}");
            assert!(!supports_explain_plan(database_type), "{database_type:?}");
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type,
                    format: None,
                    analyze: None,
                    sql: "SELECT 1".to_string(),
                })
                .reason
                .as_deref(),
                Some("unsupported"),
                "{database_type:?}"
            );
        }

        // Preserve the legacy fallback for callers that ask for a format without
        // first checking support; it does not enable plan capability.
        assert_eq!(estimated_plan_format(None), Json);
        assert_eq!(estimated_plan_format(Some(DatabaseType::Redis)), 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()) }
        );

        for sql in [
            "DROP TABLE users;",
            "DELETE FROM users;",
            "INSERT INTO users (id) VALUES (1);",
            "UPDATE users SET name = 'x';",
            "SELECT 1; DROP TABLE users;",
            "CREATE INDEX idx ON users (id);",
        ] {
            assert_eq!(
                build_explain_sql(ExplainSqlOptions {
                    database_type: Some(DatabaseType::Mysql),
                    format: None,
                    analyze: None,
                    sql: sql.to_string(),
                }),
                ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
                "{sql} must remain blocked by the explain safety gate"
            );
        }

        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_dollar_quoted() {
        // #9866: the body of a dollar-quoted literal is data, not SQL.
        assert_eq!(strip_sql_comments_and_literals("SELECT $$hello$$ FROM t"), "SELECT         FROM t");
        assert_eq!(strip_sql_comments_and_literals("SELECT $tag$hello$tag$ FROM t"), "SELECT         FROM t");
        // A literal body may contain quotes, comments and its own `$` without ending early.
        assert_eq!(
            strip_sql_comments_and_literals("SELECT $f$ a 'b' /* c */ $$ $f$ FROM t"),
            "SELECT                      FROM t"
        );
        // `$1` is a placeholder, not a literal opener.
        assert_eq!(strip_sql_comments_and_literals("SELECT $1, $2 FROM t"), "SELECT $1, $2 FROM t");
        // An unterminated `$$` is not a literal, so the text stays visible to the keyword scan.
        assert_eq!(strip_sql_comments_and_literals("SELECT $$hello FROM t"), "SELECT $$hello FROM t");
        // `$` continues an identifier, so `foo$tag$bar` is one word and not a literal.
        assert_eq!(strip_sql_comments_and_literals("SELECT foo$tag$bar FROM t"), "SELECT foo$tag$bar FROM t");
    }

    #[test]
    fn contains_dangerous_sql_keyword_ignores_dollar_quoted_literals() {
        assert!(!contains_dangerous_sql_keyword("SELECT $$INSERT INTO users VALUES (1)$$ FROM t"));
        assert!(!contains_dangerous_sql_keyword("SELECT $body$DROP TABLE users$body$ FROM t"));
        assert!(!contains_dangerous_sql_keyword("SELECT $$it''s a DELETE$$ FROM t"));
        // Nested dollar quotes in a tagged literal must not close it early.
        assert!(!contains_dangerous_sql_keyword("SELECT $outer$$DELETE FROM t$$$outer$ FROM t"));
        // An unterminated literal is malformed input and still fails closed.
        assert!(contains_dangerous_sql_keyword("SELECT $$DELETE FROM users"));
    }

    #[test]
    fn is_write_sql_dollar_quoted_literals_stay_read_only() {
        for database_type in [DatabaseType::Postgres, DatabaseType::Mysql, DatabaseType::SqlServer] {
            assert!(
                !is_write_sql_for_database("SELECT $$INSERT INTO t VALUES(1)$$ FROM t", database_type),
                "dollar-quoted literal must stay read-only for {database_type:?}"
            );
        }
        assert!(!is_write_sql("SELECT $$DROP TABLE users$$ FROM t"));
        // `SELECT ... INTO` still creates a table, so the write must survive.
        assert!(is_write_sql_for_database("SELECT $$x$$ INTO backup FROM t", DatabaseType::Postgres));
        assert!(is_write_sql_for_database("SELECT $$x$$ INTO backup FROM t", DatabaseType::SqlServer));
        // A write after a literal is unaffected.
        assert!(is_write_sql_for_database("SELECT $$x$$ FROM t; DELETE FROM t", DatabaseType::Postgres));
    }

    #[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_oracle_alternative_quoted() {
        // The quote inside q'...' is literal data, so the rest of the statement
        // stays visible to the keyword scan instead of being swallowed by the
        // mis-paired ordinary string.
        let visible = strip_sql_comments_and_literals("SELECT q'[O'Reilly]' FROM t; DELETE FROM users;");
        assert!(visible.contains("DELETE FROM users"), "{visible}");
        let visible = strip_sql_comments_and_literals("SELECT nq'{O'Reilly}' FROM t; DELETE FROM users;");
        assert!(visible.contains("DELETE FROM users"), "{visible}");
        // SQL-looking text inside the literal stays literal data.
        let hidden = strip_sql_comments_and_literals("SELECT q'[it's; DELETE FROM users;]' FROM t;");
        assert!(!hidden.contains("DELETE"), "{hidden}");
        // The introducer must start its own token: `aq` is one identifier, so
        // the following literal is ordinary.
        let visible = strip_sql_comments_and_literals("SELECT aq'[x]' FROM t; DELETE FROM users;");
        assert!(visible.contains("DELETE FROM users"), "{visible}");
        // An unterminated q-literal is malformed input and stays visible, so
        // the keyword scan fails closed.
        let visible = strip_sql_comments_and_literals("SELECT q'[O'Reilly FROM t; DELETE FROM users;");
        assert!(visible.contains("DELETE FROM users"), "{visible}");
    }

    #[test]
    fn oracle_proven_read_only_statement_alternative_quoted() {
        // The mis-paired quotes previously reduced this to `SELECT q`, proving
        // the statement read-only even though real code follows the literal.
        assert!(!is_oracle_proven_read_only_statement("SELECT q'[O'Reilly]' FROM users; DELETE FROM users;"));
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users WHERE name = q'[O''Reilly]'"));
        // SQL-looking text inside the literal is data, not a forbidden keyword.
        assert!(is_oracle_proven_read_only_statement("SELECT * FROM users WHERE note = q'[drop table]'"));
    }

    #[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_dynamodb_partiql_reads_and_writes_for_read_only_protection() {
        assert!(!is_write_sql_for_database(
            "SELECT * FROM \"items\" WHERE \"status\"='PENDING'",
            DatabaseType::DynamoDb
        ));
        assert!(!is_write_sql_for_database("-- comment\nSELECT 'UPDATE' FROM \"items\"", DatabaseType::DynamoDb));
        for statement in [
            "INSERT INTO \"items\" VALUE {'pk':'x'}",
            "UPDATE \"items\" SET flag=true WHERE pk='x'",
            "DELETE FROM \"items\" WHERE pk='x'",
            "SELECT * FROM \"items\"; DELETE FROM \"items\" WHERE pk='x'",
        ] {
            assert!(is_write_sql_for_database(statement, DatabaseType::DynamoDb), "{statement}");
        }
    }

    #[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("GET /_all_dbs", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("POST /mydb/_find\n{\"selector\":{}}", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk(
                "POST /mydb/_bulk_get\n{\"docs\":[{\"id\":\"a\"}]}",
                DatabaseType::CouchDb
            ),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk(
                "POST /mydb/_design/users/_view/by_name\n{\"keys\":[\"a\"]}",
                DatabaseType::CouchDb
            ),
            Some(SearchEngineQueryRisk::ReadOnly)
        );
        assert_eq!(
            classify_search_engine_query_risk("POST /mydb/_design/users/_info", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Write)
        );
        assert_eq!(
            classify_search_engine_query_risk("POST /mydb\n{\"name\":\"doc\"}", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Write)
        );
        assert_eq!(
            classify_search_engine_query_risk("PUT /mydb/doc1\n{\"name\":\"doc\"}", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Write)
        );
        assert_eq!(
            classify_search_engine_query_risk("PUT /mydb", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk("DELETE /mydb", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Dangerous)
        );
        assert_eq!(
            classify_search_engine_query_risk("DELETE /mydb/doc1", DatabaseType::CouchDb),
            Some(SearchEngineQueryRisk::Write)
        );
        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 classify_salesforce_statement_risk_scoped_dml() {
        let update = "DBX SALESFORCE DML\n{\"op\":\"update\",\"object\":\"Account\",\"id\":\"001x\",\"fields\":{\"Name\":\"Acme\"}}";
        assert_eq!(classify_salesforce_statement_risk(update), SalesforceStatementRisk::ScopedWrite);
        let delete = "DBX SALESFORCE DML\n{\"op\":\"delete\",\"object\":\"Account\",\"id\":\"001x\"}";
        assert_eq!(classify_salesforce_statement_risk(delete), SalesforceStatementRisk::ScopedWrite);
        let insert = "DBX SALESFORCE DML\n{\"op\":\"insert\",\"object\":\"Lead\",\"fields\":{\"Company\":\"Acme\"}}";
        assert_eq!(classify_salesforce_statement_risk(insert), SalesforceStatementRisk::ScopedWrite);
    }

    #[test]
    fn classify_salesforce_statement_risk_unscoped_dml_fails_closed() {
        // No Id: nothing bounds the write, so it is not a safe write.
        assert_eq!(
            classify_salesforce_statement_risk(
                "DBX SALESFORCE DML\n{\"op\":\"update\",\"object\":\"Account\",\"fields\":{}}"
            ),
            SalesforceStatementRisk::OpaqueWrite
        );
        // Unknown op / unknown header / unparseable body all fail closed.
        assert_eq!(
            classify_salesforce_statement_risk(
                "DBX SALESFORCE DML\n{\"op\":\"upsert\",\"object\":\"Account\",\"id\":\"001x\"}"
            ),
            SalesforceStatementRisk::OpaqueWrite
        );
        assert_eq!(
            classify_salesforce_statement_risk("DBX SALESFORCE FUTURE\n{}"),
            SalesforceStatementRisk::OpaqueWrite
        );
        assert_eq!(
            classify_salesforce_statement_risk("DBX SALESFORCE DML\n{not json"),
            SalesforceStatementRisk::OpaqueWrite
        );
    }

    #[test]
    fn classify_salesforce_statement_risk_dml_body_survives_comment_lookalikes() {
        // A `--` inside a field value must not be stripped as a line comment:
        // the pseudo-command branch reads the raw text, so the body still parses
        // and the write stays scoped.
        let dml = "DBX SALESFORCE DML\n{\"op\":\"update\",\"object\":\"Account\",\"id\":\"001x\",\"fields\":{\"Description\":\"a--b\"}}";
        assert_eq!(classify_salesforce_statement_risk(dml), SalesforceStatementRisk::ScopedWrite);
    }

    #[test]
    fn classify_salesforce_statement_risk_soql_is_read() {
        assert_eq!(classify_salesforce_statement_risk("SELECT Id FROM Account"), SalesforceStatementRisk::Read);
        assert_eq!(classify_salesforce_statement_risk("  select Name from Lead  "), SalesforceStatementRisk::Read);
        // SOQL forms sqlparser cannot parse must still read as reads.
        assert_eq!(
            classify_salesforce_statement_risk("SELECT FIELDS(ALL) FROM Account LIMIT 200"),
            SalesforceStatementRisk::Read
        );
        assert_eq!(
            classify_salesforce_statement_risk("SELECT Id FROM Opportunity WHERE CloseDate = LAST_N_DAYS:7"),
            SalesforceStatementRisk::Read
        );
        assert_eq!(
            classify_salesforce_statement_risk("SELECT Id, (SELECT Id FROM Contacts) FROM Account"),
            SalesforceStatementRisk::Read
        );
        assert_eq!(
            classify_salesforce_statement_risk("/* report */ SELECT Id FROM Account"),
            SalesforceStatementRisk::Read
        );
    }

    #[test]
    fn classify_salesforce_statement_risk_non_soql_is_opaque_write() {
        // The driver only accepts SELECT and DBX pseudo-commands; anything else
        // is a client error, not something to classify as a harmless read.
        assert_eq!(classify_salesforce_statement_risk("DELETE FROM Account"), SalesforceStatementRisk::OpaqueWrite);
        assert_eq!(
            classify_salesforce_statement_risk("FIND {Acme} IN ALL FIELDS"),
            SalesforceStatementRisk::OpaqueWrite
        );
        assert_eq!(classify_salesforce_statement_risk(""), SalesforceStatementRisk::OpaqueWrite);
    }

    #[test]
    fn is_write_sql_for_database_salesforce_soql_is_read() {
        assert!(!is_write_sql_for_database("SELECT Id FROM Account", DatabaseType::Salesforce));
    }

    #[test]
    fn is_write_sql_for_database_salesforce_literals_are_not_writes() {
        // The generic keyword scan would see DELETE/UPDATE here; SOQL cannot write.
        assert!(!is_write_sql_for_database(
            "SELECT Id FROM Case WHERE Subject = 'Delete request'",
            DatabaseType::Salesforce
        ));
        assert!(!is_write_sql_for_database("SELECT Update__c FROM Lead", DatabaseType::Salesforce));
    }

    #[test]
    fn is_write_sql_for_database_salesforce_dml_is_write() {
        let dml = "DBX SALESFORCE DML\n{\"op\":\"delete\",\"object\":\"Account\",\"id\":\"x\"}";
        assert!(is_write_sql_for_database(dml, DatabaseType::Salesforce));
    }

    #[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}");
        }
    }
}