Files
dbx/crates/dbx-sql/src/query_execution_sql.rs
T

2396 lines
94 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use serde::{Deserialize, Serialize};
use sqlparser::dialect::{MsSqlDialect, PostgreSqlDialect};
use sqlparser::tokenizer::{Token, Tokenizer};
use crate::models::connection::DatabaseType;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ExplainFormat {
Json,
Standard,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExplainSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
/// MySQL supports both a structured JSON plan and the traditional tabular plan.
/// Omitted formats retain the existing JSON behavior.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub format: Option<ExplainFormat>,
/// PostgreSQL and SQL Server only: run the statement and report measured
/// rows/timings. Every other engine ignores the flag.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub analyze: Option<bool>,
pub sql: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExplainSqlBuildResult {
pub ok: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub sql: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,
}
/// Payload encoding of an estimated plan produced by [`build_explain_sql`].
///
/// Callers that hand a raw plan to a consumer which cannot inspect the SQL
/// (the plugin Host API) need to know how to decode it. Keep this in sync with
/// the generated statements in [`build_explain_sql`];
/// `estimated_plan_format_matches_generated_sql` guards the pairing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EstimatedPlanFormat {
/// A JSON document returned as a single cell (PostgreSQL, MySQL, OceanBase Oracle).
Json,
/// A `ShowPlanXML` document returned as text (SQL Server `SET SHOWPLAN_XML`).
Xml,
/// A human-readable plan listing (Dameng, Doris, QuestDB, Oracle).
Text,
}
impl EstimatedPlanFormat {
pub const fn as_str(self) -> &'static str {
match self {
Self::Json => "json",
Self::Xml => "xml",
Self::Text => "text",
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DroppedFilePreviewSqlOptions {
pub path: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub limit: Option<usize>,
}
pub fn build_explain_sql(options: ExplainSqlOptions) -> ExplainSqlBuildResult {
if !supports_explain_plan(options.database_type) {
return explain_err("unsupported");
}
let source = strip_trailing_semicolons(options.sql.trim());
if source.is_empty() {
return explain_err("empty");
}
if !is_safe_explain_sql_for_database(&source, options.database_type) {
return explain_err("unsafe");
}
if options.analyze == Some(true)
&& options.database_type.is_some_and(|database_type| {
matches!(database_type, DatabaseType::Postgres | DatabaseType::SqlServer)
&& is_write_sql_for_database(&source, database_type)
})
{
return explain_err("unsafe");
}
if options.database_type == Some(DatabaseType::SqlServer) && crate::sql::split_sql_batches(&source).len() != 1 {
return explain_err("unsafe");
}
let sql = match options.database_type {
// ANALYZE executes the statement; is_safe_explain_sql has already limited
// the source to SELECT/WITH/TABLE/VALUES. MongoDb shares the plain arm only.
Some(DatabaseType::Postgres) if options.analyze == Some(true) => {
format!("EXPLAIN (ANALYZE, FORMAT JSON) {source}")
}
Some(DatabaseType::Postgres | DatabaseType::MongoDb) => {
format!("EXPLAIN (FORMAT JSON) {source}")
}
Some(DatabaseType::Dameng | DatabaseType::Questdb) => {
format!("EXPLAIN {source}")
}
Some(DatabaseType::Doris) => format!("EXPLAIN {source}"),
Some(DatabaseType::Oracle) => format!("EXPLAIN PLAN FOR {source}"),
// STATISTICS XML returns the same ShowPlanXML document plus per-operator
// runtime counters, at the price of actually running the statement.
Some(DatabaseType::SqlServer) if options.analyze == Some(true) => {
format!("SET STATISTICS XML ON;\nGO\n{source}\nGO\nSET STATISTICS XML OFF;")
}
Some(DatabaseType::SqlServer) => {
format!("SET SHOWPLAN_XML ON;\nGO\n{source}\nGO\nSET SHOWPLAN_XML OFF;")
}
Some(DatabaseType::Mysql) if options.format == Some(ExplainFormat::Standard) => {
// MySQL 8.0.32+ may otherwise inherit TREE or JSON from the session-level explain_format.
format!("EXPLAIN FORMAT=TRADITIONAL {source}")
}
_ => format!("EXPLAIN FORMAT=JSON {source}"),
};
ExplainSqlBuildResult { ok: true, sql: Some(sql), reason: None }
}
pub fn build_dropped_file_preview_sql(options: DroppedFilePreviewSqlOptions) -> Option<String> {
let lower = options.path.to_lowercase();
let escaped = options.path.replace('\'', "''");
let limit = options.limit.unwrap_or(1000).max(1);
if lower.ends_with(".parquet") {
return Some(format!("SELECT * FROM read_parquet('{escaped}') LIMIT {limit}"));
}
if lower.ends_with(".csv") {
return Some(format!("SELECT * FROM read_csv('{escaped}') LIMIT {limit}"));
}
if lower.ends_with(".tsv") {
return Some(format!("SELECT * FROM read_csv('{escaped}', delim='\\t') LIMIT {limit}"));
}
if lower.ends_with(".json") {
return Some(format!("SELECT * FROM read_json('{escaped}') LIMIT {limit}"));
}
None
}
/// How a consumer should decode the estimated plan `build_explain_sql` asks
/// the server for. Dameng and Oracle plans reach DBX through the driver's
/// native plan text rather than a generated `EXPLAIN` statement, and every
/// agent plan is text.
pub fn estimated_plan_format(database_type: Option<DatabaseType>) -> EstimatedPlanFormat {
match database_type {
Some(DatabaseType::SqlServer) => EstimatedPlanFormat::Xml,
Some(DatabaseType::Dameng | DatabaseType::Questdb | DatabaseType::Doris | DatabaseType::Oracle) => {
EstimatedPlanFormat::Text
}
_ => EstimatedPlanFormat::Json,
}
}
pub fn supports_explain_plan(database_type: Option<DatabaseType>) -> bool {
matches!(
database_type,
Some(
DatabaseType::Mysql
| DatabaseType::Doris
| DatabaseType::Postgres
| DatabaseType::Questdb
| DatabaseType::Dameng
| DatabaseType::Oracle
| DatabaseType::OceanbaseOracle
| DatabaseType::SqlServer
)
)
}
pub fn is_safe_explain_sql(sql: &str) -> bool {
is_safe_explain_sql_for_database(sql, None)
}
pub fn is_safe_explain_sql_for_database(sql: &str, database_type: Option<DatabaseType>) -> bool {
let source = strip_trailing_semicolons(sql.trim());
if source.is_empty() || has_extra_statement_after_semicolon(&source) {
return false;
}
if database_type == Some(DatabaseType::Oracle) && is_safe_oracle_explain_dml_source(&source) {
return true;
}
is_safe_explain_source(&source) && !contains_dangerous_sql_keyword(&source)
}
/// Returns true for databases that support SQL query execution (execute_query / get_sample_data).
/// Non-SQL databases (Redis, MongoDB, Elasticsearch, InfluxDB, VictoriaMetrics, Neo4j, etcd) are excluded.
pub fn supports_sql_query(database_type: DatabaseType) -> bool {
!matches!(
database_type,
DatabaseType::Redis
| DatabaseType::MongoDb
| DatabaseType::Elasticsearch
| DatabaseType::Easysearch
| DatabaseType::Solr
| DatabaseType::Qdrant
| DatabaseType::Milvus
| DatabaseType::Weaviate
| DatabaseType::ChromaDb
| DatabaseType::InfluxDb
| DatabaseType::InfluxDb3
| DatabaseType::VictoriaMetrics
| DatabaseType::Neo4j
| DatabaseType::Etcd
)
}
pub fn is_safe_dameng_autotrace_sql(sql: &str) -> bool {
let source = strip_trailing_semicolons(sql.trim());
if source.is_empty() || has_extra_statement_after_semicolon(&source) {
return false;
}
is_safe_explain_source(&source) && !contains_dangerous_sql_keyword(&source)
}
fn explain_err(reason: &str) -> ExplainSqlBuildResult {
ExplainSqlBuildResult { ok: false, sql: None, reason: Some(reason.to_string()) }
}
fn strip_trailing_semicolons(sql: &str) -> String {
sql.trim_end().trim_end_matches(';').trim_end().to_string()
}
fn is_safe_explain_source(sql: &str) -> bool {
let source = strip_sql_comments(sql).trim_start().to_lowercase();
["select", "with", "table", "values"].iter().any(|keyword| starts_with_explain_keyword_boundary(&source, keyword))
}
fn starts_with_explain_keyword_boundary(source: &str, keyword: &str) -> bool {
let Some(remainder) = source.strip_prefix(keyword) else {
return false;
};
match remainder.chars().next() {
None => true,
Some(character) if character.is_ascii_alphanumeric() || matches!(character, '_' | '$') => false,
// The safety gate permits ASCII whitespace and punctuation after the
// leading keyword, but fails closed for non-ASCII continuation chars.
Some(character) if character.is_ascii() => true,
Some(_) => false,
}
}
fn is_safe_oracle_explain_dml_source(sql: &str) -> bool {
let source = strip_sql_comments_and_literals(sql);
let source = source.trim_start().to_ascii_uppercase();
["INSERT", "UPDATE", "DELETE", "MERGE"].iter().any(|keyword| starts_with_keyword(&source, keyword))
}
/// Write verbs that mark a read-leading statement as a write, for example a
/// data-modifying CTE (`WITH x AS (DELETE FROM t RETURNING *) SELECT * FROM x`).
const DANGEROUS_SQL_KEYWORDS: &[&str] =
&["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"];
/// Write verbs that are also scalar function names, so `REPLACE(col, 'a', 'b')`
/// and `INSERT(str, pos, len, new)` are ordinary parts of a read-only query.
///
/// Only these two verbs are exempted when they are called with parentheses.
/// `update` stays dangerous even then: MySQL accepts a parenthesized table
/// reference (`UPDATE (t) SET c = 1`, verified on MySQL 8.4), while
/// `REPLACE (`/`INSERT (` are rejected there as statements, so a parenthesized
/// call can never hide a write for the exempted verbs.
const WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS: &[&str] = &["replace", "insert"];
pub fn contains_dangerous_sql_keyword(sql: &str) -> bool {
let source = strip_sql_comments_and_literals(sql).to_lowercase();
DANGEROUS_SQL_KEYWORDS.iter().any(|keyword| contains_dangerous_sql_verb(&source, keyword))
}
/// Match a write verb as a whole word. A verb called with parentheses is a
/// scalar function for [`WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS`] (`SELECT
/// REPLACE(name, 'a', 'b') FROM users` must stay read-only), including when the
/// dialect allows whitespace before the parenthesis.
fn contains_dangerous_sql_verb(source: &str, keyword: &str) -> bool {
let bytes = source.as_bytes();
let keyword_bytes = keyword.as_bytes();
if keyword_bytes.is_empty() || bytes.len() < keyword_bytes.len() {
return false;
}
let allows_function_form = WRITE_VERBS_WITH_SCALAR_FUNCTION_FORMS.contains(&keyword);
for idx in 0..=bytes.len() - keyword_bytes.len() {
if &bytes[idx..idx + keyword_bytes.len()] != keyword_bytes {
continue;
}
if is_identifier_byte(idx.checked_sub(1).and_then(|i| bytes.get(i)).copied()) {
continue;
}
let rest = &bytes[idx + keyword_bytes.len()..];
match rest.first().copied() {
None => return true,
Some(b'(') if allows_function_form => continue,
Some(byte) if is_identifier_byte(Some(byte)) => continue,
Some(byte) => {
if allows_function_form && byte.is_ascii_whitespace() && starts_with_parenthesis(rest) {
continue;
}
return true;
}
}
}
false
}
/// True when the first non-whitespace byte is `(` (SQL Server style `REPLACE (a, b, c)`).
fn starts_with_parenthesis(rest: &[u8]) -> bool {
let offset = rest.iter().take_while(|byte| byte.is_ascii_whitespace()).count();
rest.get(offset) == Some(&b'(')
}
/// Keywords that start a read-only SQL statement.
/// Note: FROM is a DuckDB-specific read keyword supporting SELECT-less FROM syntax
/// (e.g. `FROM table SELECT *`). In other databases, a statement starting with FROM
/// is invalid and would be rejected by the database itself, so allowing it poses no risk.
///
/// PRAGMA is intentionally NOT in this list because some PRAGMA statements modify
/// database or session state (e.g. SQLite `PRAGMA journal_mode=WAL`). Instead,
/// read-only PRAGMA forms are handled separately in `is_safe_read_pragma`.
const READ_SQL_KEYWORDS: &[&str] = &["SELECT", "WITH", "SHOW", "DESCRIBE", "DESC", "EXPLAIN", "FROM"];
/// PRAGMA names that are known to be safe read-only queries in SQLite/DuckDB.
/// Only the function-call form `PRAGMA name(args)` matching these names is allowed.
/// Any PRAGMA with assignment (`PRAGMA name = value`) or not in this list is blocked.
const SAFE_READ_PRAGMA_NAMES: &[&str] = &[
"TABLE_INFO",
"TABLE_XINFO",
"INDEX_LIST",
"INDEX_INFO",
"FOREIGN_KEY_LIST",
"DATABASE_LIST",
"COMPILE_OPTIONS",
"DATA_VERSION",
];
/// Returns true if the SQL statement is a write operation (not a pure read).
///
/// Callers that know the connection database type should use
/// [`is_write_sql_for_database`] so executable comments are interpreted using
/// the correct dialect. This untyped helper deliberately remains conservative
/// for executable comments.
pub fn is_write_sql(sql: &str) -> bool {
is_write_sql_with_database_type(sql, None)
}
/// Returns true if the SQL statement is a write operation for a database
/// dialect. In addition to ordinary write statements, this recognizes MySQL
/// executable comments and file exports, plus PostgreSQL-family/SQL Server
/// `SELECT ... INTO` table creation.
pub fn is_write_sql_for_database(sql: &str, database_type: DatabaseType) -> bool {
// VictoriaMetrics execution is hard-wired to the read-only query API; MetricsQL
// does not use SQL verbs and must not be rejected by SQL write classification.
if database_type == DatabaseType::VictoriaMetrics {
return false;
}
if database_type == DatabaseType::DynamoDb {
if let Some(is_write) = classify_dynamodb_statement_write(sql) {
return is_write;
}
}
if let Some(risk) = classify_search_engine_query_risk(sql, database_type) {
return risk != SearchEngineQueryRisk::ReadOnly;
}
is_write_sql_with_database_type(sql, Some(database_type))
}
fn classify_dynamodb_statement_write(source: &str) -> Option<bool> {
let header = source.lines().find(|line| !line.trim().is_empty())?.trim().to_ascii_uppercase();
match header.as_str() {
"DBX DYNAMODB SCAN" | "DBX DYNAMODB QUERY / SCAN" => Some(false),
"DBX DYNAMODB INSERT ITEM" | "DBX DYNAMODB PUT ITEM" | "DBX DYNAMODB DELETE ITEM" => Some(true),
_ if header.starts_with("DBX DYNAMODB") => Some(true),
_ => None,
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SearchEngineQueryRisk {
ReadOnly,
Write,
Dangerous,
}
pub fn classify_search_engine_query_risk(source: &str, database_type: DatabaseType) -> Option<SearchEngineQueryRisk> {
if database_type == DatabaseType::Solr {
return classify_solr_query_risk(source);
}
if !matches!(database_type, DatabaseType::Elasticsearch | DatabaseType::Easysearch) {
return None;
}
let source = strip_leading_search_engine_comments(source);
let request_line = source.lines().next()?.trim();
let mut parts = request_line.split_whitespace();
let method = parts.next()?.to_ascii_uppercase();
let path = parts.next()?;
let path = path.split('?').next().unwrap_or(path).trim_end_matches('/');
let segments = path.split('/').filter(|segment| !segment.is_empty()).collect::<Vec<_>>();
let has_segment = |candidate: &str| segments.iter().any(|segment| segment.eq_ignore_ascii_case(candidate));
let has_document_id = |candidate: &str| {
segments
.iter()
.position(|segment| segment.eq_ignore_ascii_case(candidate))
.is_some_and(|index| segments.get(index + 1).is_some_and(|value| !value.is_empty()))
};
match method.as_str() {
"GET" | "HEAD" | "OPTIONS" => Some(SearchEngineQueryRisk::ReadOnly),
"POST"
if [
"_search",
"_count",
"_sql",
"_msearch",
"_field_caps",
"_terms_enum",
"_validate",
"_explain",
"_rank_eval",
"_search_shards",
]
.iter()
.any(|endpoint| has_segment(endpoint)) =>
{
Some(SearchEngineQueryRisk::ReadOnly)
}
"POST" if ["_doc", "_create", "_update", "_bulk"].iter().any(|endpoint| has_segment(endpoint)) => {
Some(SearchEngineQueryRisk::Write)
}
"PUT" if has_document_id("_doc") || has_document_id("_create") => Some(SearchEngineQueryRisk::Write),
"DELETE" if has_document_id("_doc") => Some(SearchEngineQueryRisk::Write),
"POST" | "PUT" | "PATCH" | "DELETE" => Some(SearchEngineQueryRisk::Dangerous),
_ => None,
}
}
/// Solr REST 写请求集中在 `/{core}/update`(含 JSON/XML/CSV update handlers);
/// 只读 POST handler 名单之外的写路径(`/admin/*`、`/schema`、`/config`、
/// collection/core 管理)一律按 Dangerous 处理。
fn classify_solr_query_risk(source: &str) -> Option<SearchEngineQueryRisk> {
const READ_ONLY_POST_ENDPOINTS: &[&str] = &[
"select",
"query",
"get",
"export",
"terms",
"suggest",
"spell",
"mlt",
"sql",
"graph",
"clustering",
"tvrh",
"luke",
"elevate",
"browse",
"debug",
];
let source = strip_leading_search_engine_comments(source);
let request_line = source.lines().next()?.trim();
let mut parts = request_line.split_whitespace();
let method = parts.next()?.to_ascii_uppercase();
let raw_path = parts.next()?;
let path = raw_path.split('?').next().unwrap_or(raw_path).trim_end_matches('/');
let mut segments: Vec<&str> = path.split('/').filter(|segment| !segment.is_empty()).collect();
if segments.first().is_some_and(|segment| segment.eq_ignore_ascii_case("solr")) {
segments.remove(0);
}
let endpoint = segments.last().copied().unwrap_or("");
// update handler 的变体(/update/json、/update/json/docs、/update/csv)在
// update 段之后还有子路径,所以不能只检查末段。
let is_update_path = segments.iter().any(|segment| segment.starts_with("update") || *segment == "commit");
match method.as_str() {
// Solr 的 update handler 也响应 GET(commit/optimize/stream.body 都能
// 改数据),所以 update 路径对任何方法都算写,不能只放行 POST。
_ if is_update_path => Some(SearchEngineQueryRisk::Write),
"GET" | "HEAD" | "OPTIONS" => {
if solr_admin_request_is_mutation(raw_path, &segments)
|| solr_replication_request_is_mutation(raw_path, &segments)
{
Some(SearchEngineQueryRisk::Dangerous)
} else {
Some(SearchEngineQueryRisk::ReadOnly)
}
}
"POST" if READ_ONLY_POST_ENDPOINTS.contains(&endpoint) => Some(SearchEngineQueryRisk::ReadOnly),
"POST" | "PUT" | "PATCH" | "DELETE" => Some(SearchEngineQueryRisk::Dangerous),
_ => None,
}
}
/// CoreAdmin/Collections 管理端点接受 GET 触发的变更动作
///(`GET /admin/cores?action=CREATE&name=x` 会真实建 core),只有显式只读的
/// action 才能算 ReadOnly;其余带 action 的一律按 Dangerous 处理。
fn solr_admin_request_is_mutation(raw_path: &str, segments: &[&str]) -> bool {
let is_admin_target = segments.windows(2).any(|pair| {
pair[0].eq_ignore_ascii_case("admin")
&& matches!(pair[1].to_ascii_lowercase().as_str(), "cores" | "collections")
});
if !is_admin_target {
return false;
}
let Some(query) = raw_path.split('?').nth(1) else { return false };
let action = query.split('&').find_map(|param| {
param.split_once('=').filter(|(key, _)| key.eq_ignore_ascii_case("action")).map(|(_, value)| value)
});
match action {
None => false,
Some(action) => !matches!(action.to_ascii_uppercase().as_str(), "STATUS" | "REQUESTSTATUS" | "LIST"),
}
}
/// ReplicationHandler 也接受 GET 触发的变更
///(`GET /{core}/replication?command=disablereplication` 会真实停用复制),
/// 只有显式只读的 command 才算 ReadOnly;不带 command 时 handler 默认返回
/// 详情,按只读处理。
fn solr_replication_request_is_mutation(raw_path: &str, segments: &[&str]) -> bool {
if !segments.last().is_some_and(|segment| segment.eq_ignore_ascii_case("replication")) {
return false;
}
const READ_ONLY_COMMANDS: &[&str] = &[
"details",
"restorestatus",
"filelist",
"filecontent",
"filedownload",
"filemtime",
"indexversion",
"showversion",
];
let Some(query) = raw_path.split('?').nth(1) else {
return false;
};
let command = query.split('&').find_map(|param| {
param.split_once('=').filter(|(key, _)| key.eq_ignore_ascii_case("command")).map(|(_, value)| value)
});
match command {
None => false,
Some(command) => !READ_ONLY_COMMANDS.iter().any(|readonly| readonly.eq_ignore_ascii_case(command)),
}
}
fn strip_leading_search_engine_comments(input: &str) -> &str {
let mut rest = input;
loop {
rest = rest.trim_start();
if let Some(comment) = rest.strip_prefix('#').or_else(|| rest.strip_prefix("//")) {
rest = comment.split_once('\n').map_or("", |(_, remaining)| remaining);
continue;
}
if let Some(comment) = rest.strip_prefix("/*") {
rest = comment.split_once("*/").map_or("", |(_, remaining)| remaining);
continue;
}
return rest.trim();
}
}
fn is_write_sql_with_database_type(sql: &str, database_type: Option<DatabaseType>) -> bool {
if database_type == Some(DatabaseType::SqlServer) && is_sqlserver_plan_capture_set(sql) {
return false;
}
if database_type.is_some_and(|database_type| has_dialect_specific_write(sql, database_type)) {
return true;
}
// The untyped helper remains conservative for MySQL executable comments.
// Typed callers handle those comments in has_dialect_specific_write above.
let detect_mysql_executable_comments = database_type.is_none();
let detect_select_into = database_type.is_none();
let statements = match database_type {
Some(database_type) => crate::sql::split_sql_statements_for_database(sql, database_type),
None => crate::sql::split_sql_statements(sql),
};
let allow_mysql_checksum = database_type == Some(DatabaseType::Mysql);
statements.iter().any(|statement| {
is_write_sql_statement(statement, detect_mysql_executable_comments, detect_select_into, allow_mysql_checksum)
})
}
/// The explain flows toggle plan capture with a standalone SET statement:
/// `SHOWPLAN_XML` for the estimated plan, `STATISTICS XML` for the actual one.
fn is_sqlserver_plan_capture_set(sql: &str) -> bool {
let normalized = strip_sql_comments(sql)
.split_whitespace()
.map(|part| part.trim_end_matches(';').to_ascii_uppercase())
.collect::<Vec<_>>();
let tokens = normalized.iter().map(String::as_str).collect::<Vec<_>>();
matches!(tokens.as_slice(), ["SET", "SHOWPLAN_XML", "ON" | "OFF"] | ["SET", "STATISTICS", "XML", "ON" | "OFF"])
}
fn is_mysql_compatible_database(database_type: DatabaseType) -> bool {
matches!(
database_type,
DatabaseType::Mysql
| DatabaseType::Doris
| DatabaseType::StarRocks
| DatabaseType::ManticoreSearch
| DatabaseType::Goldendb
)
}
fn is_postgresql_family_database(database_type: DatabaseType) -> bool {
matches!(
database_type,
DatabaseType::Postgres
| DatabaseType::Redshift
| DatabaseType::Gaussdb
| DatabaseType::OpenGauss
| DatabaseType::Kingbase
| DatabaseType::Highgo
| DatabaseType::Uxdb
| DatabaseType::Vastbase
| DatabaseType::Kwdb
)
}
/// Detects write-capable syntax that otherwise looks like a read query and is
/// interpreted differently depending on the database dialect.
pub fn has_dialect_specific_write(sql: &str, database_type: DatabaseType) -> bool {
let statements = crate::sql::split_sql_statements_for_database(sql, database_type);
statements.iter().any(|statement| has_dialect_specific_write_statement(statement, database_type))
}
fn has_dialect_specific_write_statement(sql: &str, database_type: DatabaseType) -> bool {
if is_mysql_compatible_database(database_type) {
let (cleaned, has_executable_comment) = strip_sql_comments_and_literals_with_metadata(sql, true);
return has_executable_comment
|| contains_keyword_sequence(&cleaned, "INTO", "OUTFILE")
|| contains_keyword_sequence(&cleaned, "INTO", "DUMPFILE");
}
if is_postgresql_family_database(database_type) {
contains_unquoted_keyword(sql, &PostgreSqlDialect {}, "INTO")
} else {
match database_type {
DatabaseType::SqlServer => contains_unquoted_keyword(sql, &MsSqlDialect {}, "INTO"),
_ => false,
}
}
}
fn contains_keyword_sequence(sql: &str, first: &str, second: &str) -> bool {
let words = sql.split(|ch: char| !ch.is_ascii_alphanumeric() && ch != '_').filter(|word| !word.is_empty());
let mut previous_matches = false;
for word in words {
if previous_matches && word.eq_ignore_ascii_case(second) {
return true;
}
previous_matches = word.eq_ignore_ascii_case(first);
}
false
}
pub fn contains_unquoted_keyword(sql: &str, dialect: &dyn sqlparser::dialect::Dialect, keyword: &str) -> bool {
Tokenizer::new(dialect, sql).tokenize().is_ok_and(|tokens| {
tokens.into_iter().any(|token| {
matches!(token, Token::Word(word) if word.quote_style.is_none() && word.value.eq_ignore_ascii_case(keyword))
})
})
}
fn is_write_sql_statement(
sql: &str,
detect_mysql_executable_comments: bool,
detect_select_into: bool,
allow_mysql_checksum: bool,
) -> bool {
// 1. Strip comments and string literals
let (cleaned, has_mysql_executable_comment) =
strip_sql_comments_and_literals_with_metadata(sql, detect_mysql_executable_comments);
// MySQL/MariaDB executable comments may contain arbitrary SQL, including
// writes that are not represented by the outer statement (for example,
// INTO OUTFILE inside a SELECT). Treat them as writes rather than
// attempting to parse every supported MySQL dialect extension here.
if has_mysql_executable_comment {
return true;
}
let trimmed = cleaned.trim_start();
if trimmed.is_empty() {
return false;
}
let upper = trimmed.to_uppercase();
// 2. Check if first keyword is a read keyword
let starts_with_read = READ_SQL_KEYWORDS.iter().any(|kw| {
upper.starts_with(kw) && (upper.len() == kw.len() || !upper.as_bytes()[kw.len()].is_ascii_alphanumeric())
}) || (allow_mysql_checksum && starts_with_keyword(&upper, "CHECKSUM TABLE"));
// 3. Special handling for PRAGMA: only allow safe read-only forms
if !starts_with_read && starts_with_keyword(&upper, "PRAGMA") {
return !is_safe_read_pragma(&upper);
}
// SHOW CREATE returns object metadata; CREATE is part of its read-only syntax.
if starts_with_show_create(&upper) {
return false;
}
// Untyped callers stay conservative. Typed callers handle SELECT ... INTO
// only for database families where the syntax performs a write.
if detect_select_into && starts_with_keyword(&upper, "SELECT") && select_contains_top_level_into(&upper) {
return true;
}
// A statement is a write if it doesn't start with a read keyword,
// or if it contains embedded dangerous keywords (e.g. CTE-wrapped writes like WITH ... AS (DELETE FROM ...))
!starts_with_read || contains_dangerous_sql_keyword(sql)
}
fn starts_with_show_create(upper: &str) -> bool {
let Some(after_show) = upper.strip_prefix("SHOW") else {
return false;
};
if !after_show.is_empty() && after_show.as_bytes()[0].is_ascii_alphanumeric() {
return false;
}
starts_with_keyword(after_show.trim_start(), "CREATE")
}
fn select_contains_top_level_into(upper: &str) -> bool {
let mut token = String::new();
let mut depth = 0usize;
let mut saw_select = false;
for ch in upper.chars().chain(std::iter::once(' ')) {
if ch.is_ascii_alphanumeric() || ch == '_' {
token.push(ch);
continue;
}
if !token.is_empty() {
if depth == 0 {
if token == "SELECT" {
saw_select = true;
} else if saw_select && token == "INTO" {
return true;
}
}
token.clear();
}
if ch == '(' {
depth += 1;
} else if ch == ')' {
depth = depth.saturating_sub(1);
}
}
false
}
/// Check if a PRAGMA statement is a safe read-only form.
/// Allows: PRAGMA table_info(...), PRAGMA index_list(...), etc.
/// Blocks: PRAGMA name = value, PRAGMA name(value), or unknown PRAGMA names.
fn is_safe_read_pragma(upper_stripped: &str) -> bool {
// Skip "PRAGMA" keyword to get the rest
let rest = upper_stripped.strip_prefix("PRAGMA").unwrap_or("").trim_start();
if rest.is_empty() {
return false;
}
// Extract the pragma name (first word)
let name_end = rest.find(|c: char| !c.is_ascii_alphanumeric() && c != '_').unwrap_or(rest.len());
let pragma_name = &rest[..name_end];
// Check if it's in the safe list
if !SAFE_READ_PRAGMA_NAMES.contains(&pragma_name) {
return false;
}
// Check the form after the name: must be function-call style "(...)" or end of statement
let after_name = rest[name_end..].trim_start();
if after_name.is_empty() {
// PRAGMA table_info (no args) — safe
return true;
}
if after_name.starts_with('(') {
// PRAGMA table_info(users) — safe read form
return true;
}
// PRAGMA table_info = something or other unsafe form — blocked
false
}
fn starts_with_keyword(upper: &str, keyword: &str) -> bool {
upper.starts_with(keyword)
&& (upper.len() == keyword.len() || !upper.as_bytes()[keyword.len()].is_ascii_alphanumeric())
}
/// Whether one MySQL statement is an explicit transaction opener that a
/// tab-scoped auto-commit session may keep alive across executions: a bare
/// `BEGIN` / `BEGIN WORK` / `START TRANSACTION [modifiers]`.
///
/// `COMMIT` / `ROLLBACK` close a transaction instead of opening one, and
/// compound statements (`BEGIN ... END`, only valid inside stored programs)
/// never open a transaction at the top level, so neither is matched. Anything
/// ambiguous — extra statements in the same text, other `BEGIN` continuations —
/// is reported as "not an opener" so the caller falls back to the historical
/// cleanup instead of keeping a transaction open by accident.
pub fn mysql_statement_opens_explicit_transaction(sql: &str) -> bool {
let cleaned = strip_sql_comments(sql);
let mut parts = cleaned.split(';');
let statement = parts.next().unwrap_or_default().trim();
if statement.is_empty() || parts.any(|part| !part.trim().is_empty()) {
return false;
}
let upper = statement.to_ascii_uppercase();
let mut tokens = upper.split_whitespace();
match tokens.next() {
Some("BEGIN") => matches!(tokens.next(), None | Some("WORK")),
// Modifiers (`READ ONLY`, `READ WRITE`, `WITH CONSISTENT SNAPSHOT`) are
// optional and are verified by the server; a malformed one fails the
// statement, so no transaction is left open for this execution.
Some("START") => tokens.next() == Some("TRANSACTION"),
_ => false,
}
}
/// Check whether a SQL statement is allowed under read-only mode.
/// Returns Err with a descriptive message if the statement is a write operation.
pub fn check_read_only(sql: &str, connection_name: &str, database_type: DatabaseType) -> Result<(), String> {
if is_write_sql_for_database(sql, database_type) {
return Err(format!(
"Read-only mode: connection '{}' has read-only protection enabled. Write operation (including stored procedure calls) blocked.",
connection_name
));
}
Ok(())
}
/// Oracle-only, fail-closed classifier for manual-transaction UX. Returns true
/// only for an ordinary top-level read that DBX can prove harmless: a single
/// top-level `SELECT` (not `WITH`) that after Oracle-aware comment/literal
/// stripping contains no row-locking, transaction-control, DDL/DCL, `CALL`/
/// `EXEC`, PL/SQL block, `EXECUTE IMMEDIATE`, database-link, sequence
/// pseudocolumn (`NEXTVAL`/`CURRVAL`), function-invocation, or subquery
/// syntax. Any lexical ambiguity is **not** proven read-only; the classifier
/// deliberately whitelists no functions because user/package function side
/// effects cannot be proven and would create an unbounded compatibility
/// surface.
pub fn is_oracle_proven_read_only_statement(sql: &str) -> bool {
let cleaned = strip_sql_comments_and_literals(sql);
let trimmed = cleaned.trim_start();
if trimmed.is_empty() {
return false;
}
let upper = trimmed.to_uppercase();
// Single top-level SELECT only (WITH is excluded because data-modifying
// CTEs can hide writes in a row-returning path).
if !starts_with_keyword(&upper, "SELECT") {
return false;
}
let lower = cleaned.to_lowercase();
// Function invocation or subquery syntax: any '(' after literal stripping.
// This deliberately excludes SELECT upper(name), COUNT(*), package
// functions, and subqueries DBX cannot prove harmless.
if lower.contains('(') {
return false;
}
// Database link: '@' cannot appear in an unquoted Oracle identifier, so any
// remaining '@' after literal stripping is a remote-object reference.
if lower.contains('@') {
return false;
}
// Word-boundary forbidden keywords. These are Oracle keywords/reserved
// words, so they cannot be legitimate unquoted identifiers inside a simple
// read; a quoted occurrence is already stripped to spaces above.
const FORBIDDEN_WORDS: &[&str] = &[
// row locking
"for",
"update",
"lock",
// transaction control
"begin",
"commit",
"rollback",
"savepoint",
// DDL/DCL
"create",
"alter",
"drop",
"truncate",
"rename",
"grant",
"revoke",
"comment",
"purge",
"audit",
"noaudit",
"analyze",
"flashback",
"merge",
"insert",
"delete",
// set operators (a UNION is not a single top-level SELECT)
"union",
"minus",
"intersect",
// stored procedures / dynamic SQL / PL/SQL blocks
"call",
"exec",
"execute",
"declare",
// sequence pseudocolumns
"nextval",
"currval",
// SELECT ... INTO is PL/SQL-only in Oracle
"into",
];
!FORBIDDEN_WORDS.iter().any(|word| contains_word(&lower, word))
}
fn contains_word(source: &str, word: &str) -> bool {
let bytes = source.as_bytes();
let word_bytes = word.as_bytes();
if word_bytes.is_empty() || bytes.len() < word_bytes.len() {
return false;
}
for idx in 0..=bytes.len() - word_bytes.len() {
if &bytes[idx..idx + word_bytes.len()] != word_bytes {
continue;
}
let before = idx.checked_sub(1).and_then(|i| bytes.get(i)).copied();
let after = bytes.get(idx + word_bytes.len()).copied();
if !is_identifier_byte(before) && !is_identifier_byte(after) {
return true;
}
}
false
}
fn is_identifier_byte(byte: Option<u8>) -> bool {
byte.is_some_and(|b| b.is_ascii_alphanumeric() || b == b'_')
}
fn has_extra_statement_after_semicolon(sql: &str) -> bool {
let stripped = strip_sql_comments_and_literals(sql);
stripped.split(';').skip(1).any(|part| !part.trim().is_empty())
}
pub fn strip_sql_comments(sql: &str) -> String {
let mut output = String::with_capacity(sql.len());
let mut chars = sql.chars().peekable();
let mut in_line_comment = false;
let mut in_block_comment = false;
while let Some(ch) = chars.next() {
if in_line_comment {
if ch == '\n' {
in_line_comment = false;
output.push(' ');
}
continue;
}
if in_block_comment {
if ch == '*' && chars.peek() == Some(&'/') {
chars.next();
in_block_comment = false;
output.push(' ');
}
continue;
}
if ch == '-' && chars.peek() == Some(&'-') {
chars.next();
in_line_comment = true;
continue;
}
if ch == '#' {
in_line_comment = true;
continue;
}
if ch == '/' && chars.peek() == Some(&'*') {
chars.next();
if let Some(body) = read_mysql_executable_comment_body(&mut chars) {
output.push(' ');
output.push_str(&strip_sql_comments(&body));
output.push(' ');
} else {
in_block_comment = true;
}
continue;
}
output.push(ch);
}
output
}
pub fn strip_sql_comments_and_literals(sql: &str) -> String {
strip_sql_comments_and_literals_with_metadata(sql, false).0
}
fn strip_sql_comments_and_literals_with_metadata(sql: &str, detect_mysql_executable_comments: bool) -> (String, bool) {
let mut output = String::with_capacity(sql.len());
let mut chars = sql.chars().peekable();
let mut in_line_comment = false;
let mut in_block_comment = false;
let mut in_single_quote = false;
let mut in_double_quote = false;
let mut in_backtick_quote = false;
let mut has_mysql_executable_comment = false;
while let Some(ch) = chars.next() {
if in_line_comment {
if ch == '\n' {
in_line_comment = false;
output.push(' ');
}
continue;
}
if in_block_comment {
if ch == '*' && chars.peek() == Some(&'/') {
chars.next();
in_block_comment = false;
output.push(' ');
}
continue;
}
if in_single_quote {
if ch == '\'' {
if chars.peek() == Some(&'\'') {
chars.next();
} else {
in_single_quote = false;
}
}
output.push(' ');
continue;
}
if in_double_quote {
if ch == '"' {
if chars.peek() == Some(&'"') {
chars.next();
} else {
in_double_quote = false;
}
}
output.push(' ');
continue;
}
if in_backtick_quote {
if ch == '`' {
if chars.peek() == Some(&'`') {
chars.next();
} else {
in_backtick_quote = false;
}
}
output.push(' ');
continue;
}
if ch == '-' && chars.peek() == Some(&'-') {
chars.next();
in_line_comment = true;
continue;
}
if ch == '#' {
in_line_comment = true;
continue;
}
if ch == '/' && chars.peek() == Some(&'*') {
chars.next();
if detect_mysql_executable_comments && is_mysql_executable_comment_start(&chars) {
has_mysql_executable_comment = true;
}
in_block_comment = true;
continue;
}
if ch == '\'' {
in_single_quote = true;
output.push(' ');
continue;
}
if ch == '"' {
in_double_quote = true;
output.push(' ');
continue;
}
if ch == '`' {
in_backtick_quote = true;
output.push(' ');
continue;
}
output.push(ch);
}
(output, has_mysql_executable_comment)
}
/// `/*! ... */` is executable in MySQL, while `/*M! ... */` (optionally
/// followed by a version number) is executable in MariaDB.
fn is_mysql_executable_comment_start(chars: &std::iter::Peekable<std::str::Chars<'_>>) -> bool {
let mut marker = chars.clone();
match marker.next() {
Some('!') => true,
Some('M') => marker.next() == Some('!'),
_ => false,
}
}
fn read_mysql_executable_comment_body<I>(chars: &mut std::iter::Peekable<I>) -> Option<String>
where
I: Iterator<Item = char>,
{
let marker = chars.peek().copied()?;
if marker == '!' {
chars.next();
} else if marker == 'M' {
chars.next();
if chars.peek() != Some(&'!') {
return None;
}
chars.next();
} else {
return None;
}
let mut body = String::new();
let mut skipping_version = true;
while let Some(ch) = chars.next() {
if ch == '*' && chars.peek() == Some(&'/') {
chars.next();
return Some(body);
}
if skipping_version && (ch.is_ascii_digit() || ch.is_whitespace()) {
continue;
}
skipping_version = false;
body.push(ch);
}
Some(body)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn oracle_proven_read_only_accepts_simple_top_level_selects() {
assert!(is_oracle_proven_read_only_statement("SELECT * FROM table"));
assert!(is_oracle_proven_read_only_statement("select id, name from users"));
assert!(is_oracle_proven_read_only_statement("SELECT * FROM users WHERE id = 1"));
assert!(is_oracle_proven_read_only_statement("SELECT * FROM users ORDER BY id"));
assert!(is_oracle_proven_read_only_statement("SELECT * FROM users FETCH FIRST 10 ROWS ONLY"));
assert!(is_oracle_proven_read_only_statement("SELECT 1"));
assert!(is_oracle_proven_read_only_statement("SELECT ROWNUM FROM dual"));
assert!(is_oracle_proven_read_only_statement("SELECT * FROM users"));
// Comment/literal stripping hides keywords that are not part of the SQL.
assert!(is_oracle_proven_read_only_statement("-- comment\nSELECT * FROM users"));
assert!(is_oracle_proven_read_only_statement("SELECT 'update' FROM dual"));
assert!(is_oracle_proven_read_only_statement("SELECT \"select\" FROM t"));
assert!(is_oracle_proven_read_only_statement("SELECT * FROM user_update"));
}
#[test]
fn oracle_proven_read_only_rejects_writes_and_ddl() {
for sql in [
"UPDATE users SET name = 'x'",
"INSERT INTO users VALUES (1)",
"DELETE FROM users",
"MERGE INTO target USING source ON (a = b) WHEN MATCHED THEN UPDATE",
"CREATE TABLE t (id INT)",
"ALTER TABLE t ADD COLUMN c INT",
"DROP TABLE t",
"TRUNCATE TABLE t",
"GRANT SELECT ON t TO x",
"REVOKE SELECT ON t FROM x",
] {
assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
}
}
#[test]
fn oracle_proven_read_only_rejects_cte_procedure_and_plsql() {
for sql in [
"WITH cte AS (SELECT 1 FROM dual) SELECT * FROM cte",
"CALL my_proc()",
"EXEC my_proc",
"EXECUTE IMMEDIATE 'DROP TABLE t'",
"BEGIN UPDATE t SET x = 1; END;",
"DECLARE x INT; BEGIN NULL; END;",
] {
assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
}
}
#[test]
fn oracle_proven_read_only_rejects_locking_and_sequence_pseudocolumns() {
for sql in [
"SELECT * FROM t FOR UPDATE",
"LOCK TABLE t IN EXCLUSIVE MODE",
"SELECT seq.NEXTVAL FROM dual",
"SELECT seq.CURRVAL FROM dual",
"SELECT * FROM t@dblink",
"SELECT * FROM t WHERE id = 1 FOR UPDATE",
] {
assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
}
}
#[test]
fn oracle_proven_read_only_rejects_functions_and_subqueries() {
for sql in [
"SELECT upper(name) FROM users",
"SELECT count(*) FROM users",
"SELECT pkg.fn() FROM dual",
"SELECT * FROM users WHERE id IN (SELECT id FROM orders)",
"SELECT * FROM (SELECT * FROM users)",
"SELECT NVL(name, 'x') FROM users",
] {
assert!(!is_oracle_proven_read_only_statement(sql), "expected rejected: {sql}");
}
}
#[test]
fn oracle_proven_read_only_rejects_empty_and_ambiguous_input() {
assert!(!is_oracle_proven_read_only_statement(""));
assert!(!is_oracle_proven_read_only_statement(" "));
assert!(!is_oracle_proven_read_only_statement("-- only a comment"));
assert!(!is_oracle_proven_read_only_statement("SELECT * FROM t WHERE a IN (1, 2)"));
assert!(!is_oracle_proven_read_only_statement("SELECT * FROM t WHERE b = 'x' AND c IN (3)"));
assert!(!is_oracle_proven_read_only_statement("SELECT 1 UNION SELECT 2"));
}
#[test]
fn builds_postgres_json_explain_sql() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: None,
sql: " select * from users where id = 1; ".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN (FORMAT JSON) select * from users where id = 1".to_string()),
reason: None,
}
);
}
#[test]
fn builds_postgres_explain_sql_for_keyword_boundaries() {
for sql in [
"SELECT* FROM users",
"SELECT\t* FROM users",
"SELECT(1)",
"VALUES(1)",
"SELECT\r\n*\r\nFROM users",
"TABLE users",
] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: None,
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: true, sql: Some(format!("EXPLAIN (FORMAT JSON) {sql}")), reason: None },
"expected safe EXPLAIN source: {sql:?}"
);
}
}
#[test]
fn rejects_explain_keyword_identifier_prefixes() {
for sql in ["SELECTED", "SELECT_foo", "SELECT$foo", "SELECT1", "VALUES_foo", "VALUES1"] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: None,
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
"expected unsafe EXPLAIN source: {sql:?}"
);
}
}
#[test]
fn builds_postgres_analyze_explain_sql() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: Some(true),
sql: " select * from users where id = 1; ".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN (ANALYZE, FORMAT JSON) select * from users where id = 1".to_string()),
reason: None,
}
);
// analyze: Some(false) must behave exactly like the omitted flag.
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: Some(false),
sql: "SELECT 1".to_string(),
})
.sql,
Some("EXPLAIN (FORMAT JSON) SELECT 1".to_string())
);
}
#[test]
fn builds_sqlserver_actual_plan_explain_sql() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: Some(true),
sql: " select * from dbo.orders where id = 1; ".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some(
"SET STATISTICS XML ON;\nGO\nselect * from dbo.orders where id = 1\nGO\nSET STATISTICS XML OFF;"
.to_string()
),
reason: None,
}
);
// analyze: Some(false) must keep the estimated SHOWPLAN plan.
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: Some(false),
sql: "SELECT 1".to_string(),
})
.sql,
Some("SET SHOWPLAN_XML ON;\nGO\nSELECT 1\nGO\nSET SHOWPLAN_XML OFF;".to_string())
);
}
#[test]
fn refuses_sqlserver_analyze_on_non_select_sources() {
// STATISTICS XML runs the statement, so the safety gate stays load-bearing.
for sql in [
"delete from dbo.orders",
"update dbo.orders set name = 'x'",
"SELECT * INTO dbo.orders_copy FROM dbo.orders",
"SELECT 1\nGO\nDROP TABLE dbo.orders",
] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: Some(true),
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
"{sql}"
);
}
}
#[test]
fn ignores_analyze_for_every_engine_but_postgres_and_sqlserver() {
for (db_type, expected) in [
(DatabaseType::MongoDb, "EXPLAIN (FORMAT JSON) SELECT 1"),
(DatabaseType::Mysql, "EXPLAIN FORMAT=JSON SELECT 1"),
(DatabaseType::Dameng, "EXPLAIN SELECT 1"),
(DatabaseType::Questdb, "EXPLAIN SELECT 1"),
(DatabaseType::Oracle, "EXPLAIN PLAN FOR SELECT 1"),
] {
let analyzed = build_explain_sql(ExplainSqlOptions {
database_type: Some(db_type),
format: None,
analyze: Some(true),
sql: "SELECT 1".to_string(),
});
let plain = build_explain_sql(ExplainSqlOptions {
database_type: Some(db_type),
format: None,
analyze: None,
sql: "SELECT 1".to_string(),
});
assert_eq!(analyzed, plain, "{db_type:?} must ignore the analyze flag");
// MongoDb is rejected earlier by supports_explain_plan, so it never
// reaches the match arm it nominally shares with Postgres.
if db_type != DatabaseType::MongoDb {
assert_eq!(analyzed.sql, Some(expected.to_string()), "{db_type:?}");
} else {
assert_eq!(analyzed.reason, Some("unsupported".to_string()));
}
}
}
#[test]
fn refuses_analyze_on_non_select_sources() {
for sql in ["delete from users", "update users set name = 'x'", "SELECT 1; DROP TABLE users"] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Postgres),
format: None,
analyze: Some(true),
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
"{sql}"
);
}
}
#[test]
fn builds_dameng_explain_sql() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Dameng),
format: None,
analyze: None,
sql: "SELECT * FROM t1 WHERE id = 1".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN SELECT * FROM t1 WHERE id = 1".to_string()),
reason: None,
}
);
}
#[test]
fn builds_doris_explain_sql_without_mysql_json_format() {
assert!(supports_explain_plan(Some(DatabaseType::Doris)));
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Doris),
format: None,
analyze: None,
sql: "SELECT * FROM users;".to_string(),
}),
ExplainSqlBuildResult { ok: true, sql: Some("EXPLAIN SELECT * FROM users".to_string()), reason: None }
);
for sql in ["DELETE FROM users", "SELECT * FROM users; DROP TABLE users"] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Doris),
format: None,
analyze: None,
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
"{sql}"
);
}
}
#[test]
fn builds_oracle_explain_plan_sql() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Oracle),
format: None,
analyze: None,
sql: "WITH rows AS (SELECT 1 AS id FROM dual) SELECT * FROM rows;".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN PLAN FOR WITH rows AS (SELECT 1 AS id FROM dual) SELECT * FROM rows".to_string()),
reason: None,
}
);
}
#[test]
fn builds_oracle_explain_plan_for_update() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Oracle),
format: None,
analyze: None,
sql: "UPDATE AA_PAY_VOUCHER_TEMP t SET t.AGENCY_ID = '1';".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN PLAN FOR UPDATE AA_PAY_VOUCHER_TEMP t SET t.AGENCY_ID = '1'".to_string()),
reason: None,
}
);
}
#[test]
fn allows_oracle_explain_plan_for_dml_only() {
for (sql, expected) in [
("INSERT INTO audit_log (id) VALUES (1)", "EXPLAIN PLAN FOR INSERT INTO audit_log (id) VALUES (1)"),
("DELETE FROM audit_log WHERE id = 1", "EXPLAIN PLAN FOR DELETE FROM audit_log WHERE id = 1"),
(
"MERGE INTO target t USING source s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.name = s.name",
"EXPLAIN PLAN FOR MERGE INTO target t USING source s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.name = s.name",
),
] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Oracle),
format: None,
analyze: None,
sql: sql.to_string(),
})
.sql,
Some(expected.to_string()),
"{sql}"
);
}
for sql in [
"DROP TABLE audit_log",
"ALTER TABLE audit_log ADD created_at DATE",
"BEGIN DELETE FROM audit_log; END;",
"UPDATE audit_log SET id = 2; DROP TABLE audit_log",
] {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Oracle),
format: None,
analyze: None,
sql: sql.to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) },
"{sql}"
);
}
}
#[test]
fn estimated_plan_format_matches_generated_sql() {
// Every dialect `supports_explain_plan` advertises must declare a payload
// format that the statement `build_explain_sql` generates really produces.
for database_type in
DatabaseType::ALL.iter().copied().filter(|database_type| supports_explain_plan(Some(*database_type)))
{
let explain_sql = build_explain_sql(ExplainSqlOptions {
database_type: Some(database_type),
format: None,
analyze: None,
sql: "SELECT 1".to_string(),
})
.sql
.expect("a supported dialect always builds EXPLAIN SQL");
match estimated_plan_format(Some(database_type)) {
EstimatedPlanFormat::Json => assert!(
explain_sql.contains("FORMAT JSON") || explain_sql.contains("FORMAT=JSON"),
"{database_type:?} declares a JSON plan but generated '{explain_sql}'"
),
EstimatedPlanFormat::Xml => assert!(
explain_sql.contains("SHOWPLAN_XML"),
"{database_type:?} declares an XML plan but generated '{explain_sql}'"
),
EstimatedPlanFormat::Text => assert!(
explain_sql.starts_with("EXPLAIN ") && !explain_sql.contains("FORMAT"),
"{database_type:?} declares a text plan but generated '{explain_sql}'"
),
}
}
assert_eq!(estimated_plan_format(None), EstimatedPlanFormat::Json);
assert_eq!(estimated_plan_format(Some(DatabaseType::Dameng)), EstimatedPlanFormat::Text);
assert_eq!(estimated_plan_format(Some(DatabaseType::Questdb)), EstimatedPlanFormat::Text);
assert_eq!(estimated_plan_format(Some(DatabaseType::Doris)), EstimatedPlanFormat::Text);
assert_eq!(estimated_plan_format(Some(DatabaseType::Oracle)), EstimatedPlanFormat::Text);
assert_eq!(estimated_plan_format(Some(DatabaseType::SqlServer)), EstimatedPlanFormat::Xml);
assert_eq!(estimated_plan_format(Some(DatabaseType::Mysql)), EstimatedPlanFormat::Json);
}
#[test]
fn builds_sqlserver_showplan_xml_batches() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: None,
sql: "WITH rows AS (SELECT 1 AS id) SELECT * FROM rows;".to_string(),
});
assert_eq!(
result,
ExplainSqlBuildResult {
ok: true,
sql: Some(
"SET SHOWPLAN_XML ON;\nGO\nWITH rows AS (SELECT 1 AS id) SELECT * FROM rows\nGO\nSET SHOWPLAN_XML OFF;"
.to_string()
),
reason: None,
}
);
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: None,
sql: "SELECT 1\nGO\nSELECT 2".to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
);
assert!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::SqlServer),
format: None,
analyze: None,
sql: "SELECT 'first line\nGO\nlast line' AS text".to_string(),
})
.ok
);
}
#[test]
fn validates_dameng_autotrace_sql_safety() {
assert!(is_safe_dameng_autotrace_sql("SELECT * FROM t WHERE name = 'delete';"));
assert!(is_safe_dameng_autotrace_sql("/* comment */ WITH q AS (SELECT 1) SELECT * FROM q"));
assert!(!is_safe_dameng_autotrace_sql("SELECT * FROM t; DELETE FROM t"));
assert!(!is_safe_dameng_autotrace_sql("UPDATE t SET name = 'x'"));
assert!(!is_safe_dameng_autotrace_sql("SELECT * FROM t; /* hidden */ DROP TABLE t"));
assert!(!is_safe_dameng_autotrace_sql(""));
}
#[test]
fn builds_mysql_json_explain_and_rejects_unsafe_sql() {
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Mysql),
format: None,
analyze: None,
sql: "SELECT * FROM users;".to_string(),
}),
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN FORMAT=JSON SELECT * FROM users".to_string()),
reason: None,
}
);
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Mysql),
format: None,
analyze: None,
sql: "delete from users".to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
);
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Mysql),
format: None,
analyze: None,
sql: "SELECT * FROM users; DELETE FROM users".to_string(),
}),
ExplainSqlBuildResult { ok: false, sql: None, reason: Some("unsafe".to_string()) }
);
assert_eq!(
build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::Mysql),
format: Some(ExplainFormat::Standard),
analyze: None,
sql: "SELECT * FROM users;".to_string(),
}),
ExplainSqlBuildResult {
ok: true,
sql: Some("EXPLAIN FORMAT=TRADITIONAL SELECT * FROM users".to_string()),
reason: None,
}
);
}
#[test]
fn builds_nonexecuting_oceanbase_oracle_json_explain() {
let result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
format: Some(ExplainFormat::Json),
analyze: Some(true),
sql: "SELECT * FROM events;".to_string(),
});
assert_eq!(result.sql.as_deref(), Some("EXPLAIN FORMAT=JSON SELECT * FROM events"));
assert!(result.ok);
let unsafe_result = build_explain_sql(ExplainSqlOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
format: None,
analyze: None,
sql: "DELETE FROM events".to_string(),
});
assert_eq!(unsafe_result.reason.as_deref(), Some("unsafe"));
}
#[test]
fn builds_dropped_file_preview_sql() {
assert_eq!(
build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
path: "/tmp/O'Hara.csv".to_string(),
limit: Some(25),
}),
Some("SELECT * FROM read_csv('/tmp/O''Hara.csv') LIMIT 25".to_string())
);
assert_eq!(
build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
path: "/tmp/data.tsv".to_string(),
limit: None,
}),
Some("SELECT * FROM read_csv('/tmp/data.tsv', delim='\\t') LIMIT 1000".to_string())
);
assert_eq!(
build_dropped_file_preview_sql(DroppedFilePreviewSqlOptions {
path: "/tmp/data.txt".to_string(),
limit: None,
}),
None
);
}
#[test]
fn strip_sql_comments_and_literals_basic() {
assert_eq!(strip_sql_comments_and_literals("SELECT 1"), "SELECT 1");
assert_eq!(strip_sql_comments_and_literals("SELECT 'hello'"), "SELECT ");
assert_eq!(strip_sql_comments_and_literals("SELECT \"hello\""), "SELECT ");
assert_eq!(strip_sql_comments_and_literals("-- comment\nSELECT 1"), " SELECT 1");
assert_eq!(strip_sql_comments_and_literals("/* block */ SELECT 1"), " SELECT 1");
assert_eq!(strip_sql_comments_and_literals("# comment\nSELECT 1"), " SELECT 1");
}
#[test]
fn strip_sql_comments_and_literals_nested() {
// String literals containing comments should be stripped
assert_eq!(strip_sql_comments_and_literals("SELECT '/* not a comment */'"), "SELECT ");
// Comments containing string delimiters should be stripped
assert_eq!(strip_sql_comments_and_literals("/* 'not a string' */ SELECT 1"), " SELECT 1");
}
#[test]
fn contains_dangerous_sql_keyword_detects_writes() {
assert!(contains_dangerous_sql_keyword("DROP TABLE users"));
assert!(contains_dangerous_sql_keyword("DELETE FROM users"));
assert!(contains_dangerous_sql_keyword("TRUNCATE TABLE users"));
assert!(contains_dangerous_sql_keyword("ALTER TABLE users ADD COLUMN age INT"));
assert!(contains_dangerous_sql_keyword("UPDATE users SET name = 'x'"));
assert!(contains_dangerous_sql_keyword("MERGE INTO target USING source"));
assert!(contains_dangerous_sql_keyword("REPLACE INTO users VALUES (1)"));
assert!(contains_dangerous_sql_keyword("INSERT INTO users VALUES (1)"));
assert!(contains_dangerous_sql_keyword("CREATE TABLE users (id INT)"));
}
#[test]
fn contains_dangerous_sql_keyword_ignores_substrings() {
// "updateable" contains "update" as substring but is a different word
assert!(!contains_dangerous_sql_keyword("SELECT * FROM updateable_view"));
// "dropped" contains "drop" as substring
assert!(!contains_dangerous_sql_keyword("SELECT dropped FROM t"));
// "inserted" contains "insert"
assert!(!contains_dangerous_sql_keyword("SELECT inserted FROM t"));
}
#[test]
fn contains_dangerous_sql_keyword_ignores_scalar_functions_named_like_writes() {
// #9978: REPLACE()/INSERT() are scalar functions, not statements.
assert!(!contains_dangerous_sql_keyword("SELECT REPLACE(name, 'a', 'b') FROM users"));
assert!(!contains_dangerous_sql_keyword("SELECT REPLACE (name, 'a', 'b') FROM users"));
assert!(!contains_dangerous_sql_keyword("SELECT INSERT('Quadratic', 3, 4, 'What')"));
assert!(!contains_dangerous_sql_keyword("SELECT dropped, inserted FROM t"));
assert!(contains_dangerous_sql_keyword("REPLACE INTO users VALUES (1)"));
assert!(contains_dangerous_sql_keyword("REPLACE users SET name = 'x'"));
assert!(contains_dangerous_sql_keyword("INSERT INTO users VALUES (1)"));
// `UPDATE (t)` is a real write: MySQL accepts a parenthesized table reference.
assert!(contains_dangerous_sql_keyword("UPDATE (users) SET name = 'x'"));
}
#[test]
fn is_write_sql_allows_mysql_json_table_query_with_replace_function() {
// #9978: a JSON_TABLE query using REPLACE() must not be treated as a write,
// otherwise read-only connections ask for write permission before running it.
let sql = "with tmp as (\n select '张三,李四,王五' as `names`\n)\nSELECT jt.name\nFROM tmp t\nJOIN JSON_TABLE(\n CONCAT('[\"',REPLACE(t.`names`,',','\",\"'),'\"]'),\n '$[*]' COLUMNS(name VARCHAR(50) PATH '$')\n) jt";
assert!(!is_write_sql_for_database(sql, DatabaseType::Mysql), "JSON_TABLE + REPLACE() must stay read-only");
assert!(!is_write_sql_for_database("SELECT REPLACE(name, 'a', 'b') FROM users", DatabaseType::Mysql));
assert!(!is_write_sql_for_database("SELECT INSERT('Quadratic', 3, 4, 'What')", DatabaseType::Mysql));
assert!(!is_write_sql_for_database("SELECT REPLACE (name, 'a', 'b') FROM users", DatabaseType::SqlServer));
assert!(!is_write_sql_for_database("SELECT replace(name, 'a', 'b') FROM users", DatabaseType::Postgres));
assert!(!is_write_sql(sql));
}
#[test]
fn is_write_sql_still_blocks_writes_using_function_like_verbs() {
assert!(is_write_sql_for_database("REPLACE INTO users VALUES (1)", DatabaseType::Mysql));
assert!(is_write_sql_for_database("REPLACE users SET name = 'x'", DatabaseType::Mysql));
assert!(is_write_sql_for_database("INSERT INTO users VALUES (1)", DatabaseType::Mysql));
// MySQL accepts a parenthesized table reference, so `UPDATE (` is a write.
assert!(is_write_sql_for_database("UPDATE (users) SET name = 'x'", DatabaseType::Mysql));
assert!(is_write_sql_for_database(
"WITH x AS (SELECT 1 AS n) UPDATE (users) SET name = 'x'",
DatabaseType::Mysql
));
// Data-modifying CTEs keep being detected.
assert!(is_write_sql_for_database(
"WITH x AS (INSERT INTO users VALUES (1) RETURNING *) SELECT * FROM x",
DatabaseType::Postgres
));
assert!(is_write_sql_for_database(
"WITH x AS (DELETE FROM users RETURNING *) SELECT * FROM x",
DatabaseType::Postgres
));
}
#[test]
fn contains_dangerous_sql_keyword_ignores_in_string_literals() {
assert!(!contains_dangerous_sql_keyword("SELECT 'DROP TABLE users' FROM t"));
assert!(!contains_dangerous_sql_keyword("SELECT 'delete' FROM t"));
assert!(!contains_dangerous_sql_keyword("SELECT \"CREATE TABLE\" FROM t"));
}
#[test]
fn contains_dangerous_sql_keyword_ignores_backtick_identifiers() {
for keyword in ["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"] {
let sql = format!("SELECT 1 AS `{keyword}`");
assert!(!contains_dangerous_sql_keyword(&sql), "expected safe SQL: {sql}");
}
assert!(!contains_dangerous_sql_keyword("SELECT 1 AS `before``delete`"));
}
#[test]
fn is_write_sql_detects_simple_writes() {
assert!(is_write_sql("INSERT INTO users VALUES (1)"));
assert!(is_write_sql("UPDATE users SET name = 'x'"));
assert!(is_write_sql("DELETE FROM users"));
assert!(is_write_sql("DROP TABLE users"));
assert!(is_write_sql("CREATE TABLE users (id INT)"));
assert!(is_write_sql("ALTER TABLE users ADD COLUMN age INT"));
assert!(is_write_sql("TRUNCATE TABLE users"));
assert!(is_write_sql("EXPLAIN ANALYZE DELETE FROM users"));
assert!(is_write_sql("SELECT * INTO backup_users FROM users"));
assert!(is_write_sql("SELECT * FROM users INTO OUTFILE '/tmp/users.csv'"));
assert!(is_write_sql("COPY users FROM '/tmp/users.csv'"));
assert!(is_write_sql("/*! DELETE FROM users */"));
assert!(is_write_sql(
"MERGE INTO target USING source ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.name = s.name"
));
assert!(is_write_sql("REPLACE INTO users VALUES (1)"));
}
#[test]
fn is_write_sql_allows_reads() {
assert!(!is_write_sql("SELECT * FROM users"));
assert!(!is_write_sql("SELECT id, name FROM users WHERE active = true"));
assert!(!is_write_sql("WITH cte AS (SELECT 1) SELECT * FROM cte"));
assert!(!is_write_sql("SHOW TABLES"));
assert!(!is_write_sql("DESCRIBE users"));
assert!(!is_write_sql("DESC users"));
assert!(!is_write_sql("EXPLAIN SELECT * FROM users"));
assert!(!is_write_sql("PRAGMA table_info(users)"));
assert!(!is_write_sql("FROM users SELECT *"));
}
#[test]
fn is_write_sql_allows_backtick_identifiers_with_dangerous_keywords() {
for keyword in ["drop", "delete", "truncate", "alter", "update", "merge", "replace", "insert", "create"] {
let sql = format!("SELECT 1 AS `{keyword}`");
assert!(!is_write_sql(&sql), "expected read-only SQL: {sql}");
}
assert!(!is_write_sql("SHOW COLUMNS FROM `delete`"));
assert!(!is_write_sql("DESC `delete`"));
assert!(!is_write_sql("SELECT 1 AS `before``delete`"));
assert!(!is_write_sql("SELECT 1 AS `semi;delete`; SELECT 2"));
}
#[test]
fn is_write_sql_still_blocks_writes_with_backtick_identifiers() {
assert!(is_write_sql("DELETE FROM `users`"));
assert!(is_write_sql("DROP TABLE `users`"));
assert!(is_write_sql("UPDATE `users` SET name = 'Ada'"));
}
#[test]
fn is_write_sql_allows_show_create_statements() {
for sql in [
"SHOW CREATE TABLE users",
"show create view active_users",
"SHOW CREATE PROCEDURE refresh_users",
"SHOW CREATE FUNCTION user_count",
" /* metadata */\nSHOW\nCREATE TABLE users;",
"SHOW CREATE TABLE users; SELECT ';' AS separator",
"SHOW CREATE TABLE users /* ; DELETE FROM users */",
] {
assert!(!is_write_sql(sql), "expected read-only SQL: {sql}");
}
}
#[test]
fn is_write_sql_blocks_writes_after_show_create() {
for write_sql in [
"DROP TABLE users",
"DELETE FROM users",
"TRUNCATE TABLE users",
"ALTER TABLE users ADD COLUMN active BOOLEAN",
"UPDATE users SET active = true",
"MERGE INTO users USING source ON users.id = source.id WHEN MATCHED THEN UPDATE SET active = true",
"REPLACE INTO users VALUES (1)",
"INSERT INTO users VALUES (1)",
"CREATE TABLE audit (id INT)",
] {
let sql = format!("SHOW CREATE TABLE users; {write_sql}");
assert!(is_write_sql(&sql), "expected write SQL: {sql}");
}
}
#[test]
fn is_write_sql_ignores_leading_whitespace_and_comments() {
assert!(!is_write_sql(" /* comment */ SELECT * FROM users"));
assert!(!is_write_sql("-- comment\nSELECT * FROM users"));
assert!(is_write_sql(" /* comment */ INSERT INTO users VALUES (1)"));
}
#[test]
fn is_write_sql_blocks_mysql_and_mariadb_executable_comments() {
for sql in [
"SELECT 3156 /*! INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
"SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
"SELECT 3156 /*M! INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
"SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
] {
assert!(is_write_sql_for_database(sql, DatabaseType::Mysql), "expected write SQL: {sql}");
}
assert!(!is_write_sql_for_database(
"SELECT 3156 /* INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
DatabaseType::Mysql
));
assert!(!is_write_sql_for_database(
"SELECT '/*!50000 INTO OUTFILE \'/tmp/probe\' */' AS note",
DatabaseType::Mysql
));
}
#[test]
fn is_write_sql_treats_mysql_executable_comment_syntax_as_plain_for_other_dialects() {
for database_type in [DatabaseType::Postgres, DatabaseType::Sqlite] {
for sql in [
"SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
"SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */",
"SELECT 3156 /* ordinary block comment */",
] {
assert!(
!is_write_sql_for_database(sql, database_type),
"expected read-only SQL for {database_type:?}: {sql}"
);
}
}
}
#[test]
fn is_write_sql_blocks_dialect_specific_select_into_writes() {
for sql in [
"SELECT 3156 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'",
"SELECT 3156 INTO DUMPFILE '/var/lib/mysql-files/dbx_ro_probe.bin'",
"WITH probe AS (SELECT 3156) SELECT * FROM probe INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'",
] {
assert!(is_write_sql_for_database(sql, DatabaseType::Mysql), "expected write SQL: {sql}");
}
for database_type in [
DatabaseType::Postgres,
DatabaseType::Redshift,
DatabaseType::Gaussdb,
DatabaseType::OpenGauss,
DatabaseType::Kingbase,
DatabaseType::Highgo,
DatabaseType::Vastbase,
DatabaseType::Kwdb,
] {
for sql in [
"SELECT * INTO copied_users FROM users",
"WITH active AS (SELECT * FROM users WHERE active) SELECT * INTO active_users FROM active",
] {
assert!(
is_write_sql_for_database(sql, database_type),
"expected write SQL for {database_type:?}: {sql}"
);
}
}
for sql in [
"SELECT * INTO dbo.copied_users FROM dbo.users",
"WITH active AS (SELECT * FROM users WHERE active = 1) SELECT * INTO #active_users FROM active",
] {
assert!(is_write_sql_for_database(sql, DatabaseType::SqlServer), "expected write SQL: {sql}");
}
}
#[test]
fn is_write_sql_does_not_globally_block_into() {
for sql in [
"SELECT 3156 INTO @probe",
"SELECT 'INTO OUTFILE /tmp/probe' AS note",
"SELECT 3156 /* INTO OUTFILE '/tmp/probe' */",
"SELECT 1 AS `into`, 2 AS `outfile`",
] {
assert!(!is_write_sql_for_database(sql, DatabaseType::Mysql), "expected read-only SQL: {sql}");
}
for sql in
["SELECT 'INTO copied_users' AS note", "SELECT $$INTO copied_users$$ AS note", "SELECT 1 AS \"into\""]
{
assert!(!is_write_sql_for_database(sql, DatabaseType::Postgres), "expected read-only SQL: {sql}");
}
assert!(!is_write_sql_for_database("SELECT 1 AS [into]", DatabaseType::SqlServer));
assert!(!is_write_sql_for_database("SELECT 1 INTO unsupported", DatabaseType::Sqlite));
}
#[test]
fn is_write_sql_cte_with_nested_write() {
// CTE starting with WITH but containing a write operation inside
assert!(is_write_sql("WITH deleted AS (DELETE FROM users RETURNING id) SELECT * FROM deleted"));
assert!(is_write_sql("WITH updated AS (UPDATE users SET name = 'x' RETURNING id) SELECT * FROM updated"));
assert!(is_write_sql("WITH inserted AS (INSERT INTO users VALUES (1) RETURNING id) SELECT * FROM inserted"));
// Pure read CTE should be allowed
assert!(!is_write_sql("WITH cte AS (SELECT * FROM users) SELECT * FROM cte"));
}
#[test]
fn is_write_sql_case_insensitive() {
assert!(!is_write_sql("select * from users"));
assert!(!is_write_sql("Select * From users"));
assert!(is_write_sql("insert into users values (1)"));
assert!(is_write_sql("Insert Into users Values (1)"));
assert!(is_write_sql("update users set name = 'x'"));
}
#[test]
fn is_write_sql_edge_cases() {
assert!(!is_write_sql("")); // empty string -> not a write
assert!(!is_write_sql(" ")); // whitespace only -> not a write
assert!(is_write_sql("COMMIT")); // not a recognized read keyword
assert!(is_write_sql("ROLLBACK")); // not a recognized read keyword
assert!(is_write_sql("BEGIN")); // not a recognized read keyword
assert!(is_write_sql("GRANT SELECT ON users TO admin"));
assert!(is_write_sql("REVOKE SELECT ON users FROM admin"));
}
#[test]
fn is_write_sql_blocks_stored_procedure_calls() {
// CALL and EXEC don't start with a read keyword, so they are treated as writes
assert!(is_write_sql("CALL my_procedure(1, 2)"));
assert!(is_write_sql("call my_procedure()"));
assert!(is_write_sql("EXEC sp_update_stats"));
assert!(is_write_sql("EXECUTE sp_rename 'old', 'new'"));
assert!(is_write_sql("execute my_func()"));
}
#[test]
fn is_write_sql_allows_safe_read_pragmas() {
// Read-only PRAGMA forms (function-call style with known safe names) are allowed
assert!(!is_write_sql("PRAGMA table_info(users)"));
assert!(!is_write_sql("PRAGMA table_xinfo(users)"));
assert!(!is_write_sql("PRAGMA index_list(users)"));
assert!(!is_write_sql("PRAGMA index_info(idx_name)"));
assert!(!is_write_sql("PRAGMA foreign_key_list(users)"));
assert!(!is_write_sql("PRAGMA database_list"));
assert!(!is_write_sql("PRAGMA compile_options"));
assert!(!is_write_sql("PRAGMA data_version"));
assert!(!is_write_sql("pragma table_info(users)"));
}
#[test]
fn is_write_sql_blocks_unsafe_pragmas() {
// Assignment forms are always blocked
assert!(is_write_sql("PRAGMA journal_mode = WAL"));
assert!(is_write_sql("PRAGMA synchronous = OFF"));
assert!(is_write_sql("PRAGMA foreign_keys = ON"));
assert!(is_write_sql("PRAGMA cache_size = -2000"));
assert!(is_write_sql("PRAGMA user_version = 123"));
// Unknown PRAGMA names are blocked
assert!(is_write_sql("PRAGMA writable_schema = ON"));
assert!(is_write_sql("PRAGMA locking_mode = EXCLUSIVE"));
assert!(is_write_sql("PRAGMA temp_store = MEMORY"));
assert!(is_write_sql("PRAGMA some_unknown_pragma"));
}
#[test]
fn is_write_sql_string_literal_hides_keywords() {
// The dangerous keyword is inside a string literal, so it should NOT be detected
assert!(!is_write_sql("SELECT 'DROP TABLE users' AS hint FROM t"));
assert!(!is_write_sql("SELECT * FROM t WHERE name = 'delete'"));
}
#[test]
fn check_read_only_success_and_error() {
assert_eq!(check_read_only("SELECT * FROM users", "prod-db", DatabaseType::Mysql), Ok(()));
assert_eq!(check_read_only("WITH cte AS (SELECT 1) SELECT * FROM cte", "prod-db", DatabaseType::Mysql), Ok(()));
assert_eq!(check_read_only("SHOW CREATE TABLE users", "prod-db", DatabaseType::Mysql), Ok(()));
assert_eq!(check_read_only("SELECT 1 AS `delete`", "prod-db", DatabaseType::Mysql), Ok(()));
let err = check_read_only("DELETE FROM users", "prod-db", DatabaseType::Mysql);
assert!(err.is_err());
assert_eq!(
err.unwrap_err(),
"Read-only mode: connection 'prod-db' has read-only protection enabled. Write operation (including stored procedure calls) blocked."
);
let err2 = check_read_only("UPDATE users SET name = 'x'", "reporting-db", DatabaseType::Mysql);
assert!(err2.is_err());
assert!(err2.unwrap_err().contains("reporting-db"));
let show_create_err =
check_read_only("SHOW CREATE TABLE users; DELETE FROM users", "prod-db", DatabaseType::Mysql);
assert!(show_create_err.is_err());
assert!(show_create_err.unwrap_err().contains("Write operation"));
}
#[test]
fn check_read_only_allows_mysql_checksum_only_for_mysql() {
assert_eq!(check_read_only("CHECKSUM TABLE `issue6693_repro`", "mysql", DatabaseType::Mysql), Ok(()));
assert!(check_read_only("CHECKSUM TABLE users", "postgres", DatabaseType::Postgres).is_err());
assert!(check_read_only("CHECKSUM TABLE users", "sqlite", DatabaseType::Sqlite).is_err());
assert!(check_read_only("CHECKSUM TABLE users; DROP TABLE users", "mysql", DatabaseType::Mysql).is_err());
}
#[test]
fn check_read_only_allows_only_sqlserver_plan_capture_session_switches() {
for sql in [
"SET SHOWPLAN_XML ON;",
"-- explain\nSET SHOWPLAN_XML OFF",
"SET STATISTICS XML ON;",
"-- actual plan\nSET STATISTICS XML OFF",
] {
assert_eq!(check_read_only(sql, "readonly", DatabaseType::SqlServer), Ok(()));
}
assert!(check_read_only("SET SHOWPLAN_ALL ON", "readonly", DatabaseType::SqlServer).is_err());
assert!(check_read_only("SET STATISTICS IO ON", "readonly", DatabaseType::SqlServer).is_err());
assert!(check_read_only("SET SHOWPLAN_XML ON; SELECT 1", "readonly", DatabaseType::SqlServer).is_err());
assert!(check_read_only("SET SHOWPLAN_XML OFF; DROP TABLE users", "readonly", DatabaseType::SqlServer).is_err());
assert!(
check_read_only("SET STATISTICS XML OFF; DROP TABLE users", "readonly", DatabaseType::SqlServer).is_err()
);
}
#[test]
fn check_read_only_only_treats_executable_comments_as_writes_for_mysql_compatible_connections() {
let mysql_executable_comment = "SELECT 3156 /*!50000 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */";
let mariadb_executable_comment =
"SELECT 3156 /*M!100100 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt' */";
assert!(check_read_only(mysql_executable_comment, "mysql", DatabaseType::Mysql).is_err());
assert!(check_read_only(mariadb_executable_comment, "mariadb", DatabaseType::Mysql).is_err());
for database_type in [DatabaseType::Postgres, DatabaseType::Sqlite] {
assert_eq!(check_read_only(mysql_executable_comment, "readonly", database_type), Ok(()));
assert_eq!(check_read_only(mariadb_executable_comment, "readonly", database_type), Ok(()));
}
}
#[test]
fn check_read_only_blocks_dialect_specific_select_into_writes() {
for (sql, database_type) in [
("SELECT 3156 INTO OUTFILE '/var/lib/mysql-files/dbx_ro_probe.txt'", DatabaseType::Mysql),
("SELECT * INTO copied_users FROM users", DatabaseType::Postgres),
("SELECT * INTO copied_users FROM users", DatabaseType::Redshift),
("SELECT * INTO copied_users FROM users", DatabaseType::Gaussdb),
("SELECT * INTO copied_users FROM users", DatabaseType::OpenGauss),
("SELECT * INTO copied_users FROM users", DatabaseType::Kingbase),
("SELECT * INTO copied_users FROM users", DatabaseType::Highgo),
("SELECT * INTO copied_users FROM users", DatabaseType::Vastbase),
("SELECT * INTO copied_users FROM users", DatabaseType::Kwdb),
("SELECT * INTO #copied_users FROM users", DatabaseType::SqlServer),
] {
assert!(check_read_only(sql, "readonly", database_type).is_err(), "expected blocked SQL: {sql}");
}
}
#[test]
fn strip_sql_comments_basic() {
assert_eq!(strip_sql_comments("SELECT 1"), "SELECT 1");
assert_eq!(strip_sql_comments("-- comment\nSELECT 1"), " SELECT 1");
assert_eq!(strip_sql_comments("/* block */ SELECT 1"), " SELECT 1");
assert_eq!(strip_sql_comments("# comment\nSELECT 1"), " SELECT 1");
}
#[test]
fn strip_sql_comments_preserves_strings() {
// strip_sql_comments does NOT handle string delimiters, so it strips
// comments even inside string literals
assert_eq!(strip_sql_comments("SELECT 'hello /* not a comment */'"), "SELECT 'hello '");
}
#[test]
fn classifies_dynamodb_generated_statements_for_read_only_protection() {
assert!(!is_write_sql_for_database(
"DBX DYNAMODB SCAN\ntable: \"orders\"\nlimit: 1000",
DatabaseType::DynamoDb
));
assert!(!is_write_sql_for_database(
"DBX DYNAMODB QUERY / SCAN\ntable: \"orders\"\nfilter:\n{\"status\":\"SHIPPED\"}",
DatabaseType::DynamoDb
));
for operation in ["INSERT ITEM", "PUT ITEM", "DELETE ITEM", "UNKNOWN"] {
assert!(is_write_sql_for_database(
&format!("DBX DYNAMODB {operation}\ntable: \"orders\""),
DatabaseType::DynamoDb
));
}
}
#[test]
fn classifies_search_engine_rest_writes() {
assert!(!is_write_sql_for_database("GET /_cluster/health", DatabaseType::Easysearch));
assert!(!is_write_sql_for_database(
"POST /products/_search\n{\"query\":{\"match_all\":{}}}",
DatabaseType::Easysearch
));
assert!(is_write_sql_for_database(
"PUT /products/_doc/1?refresh=true\n{\"name\":\"Notebook\"}",
DatabaseType::Easysearch
));
}
#[test]
fn classifies_solr_rest_requests() {
// Read-only requests: select/query/get and other read handlers.
assert!(!is_write_sql_for_database("GET /mycore/select?q=*:*&rows=20", DatabaseType::Solr));
assert!(!is_write_sql_for_database(
"POST /mycore/query\n{\"query\":\"name:foo\",\"limit\":10}",
DatabaseType::Solr
));
assert!(!is_write_sql_for_database("GET /mycore/schema/fields", DatabaseType::Solr));
// Document writes hit the update handlers, including the `/update/...`
// variants whose last path segment is not literally `update`.
assert!(is_write_sql_for_database(
"POST /mycore/update\n{\"add\":{\"doc\":{\"id\":\"1\"}}}",
DatabaseType::Solr
));
assert!(is_write_sql_for_database("POST /mycore/update/json/docs\n{\"id\":\"1\"}", DatabaseType::Solr));
assert!(is_write_sql_for_database("GET /mycore/update?commit=true", DatabaseType::Solr));
// CoreAdmin actions mutate over GET too; only the read whitelist stays safe.
assert_eq!(
classify_search_engine_query_risk("GET /admin/cores?action=CREATE&name=x", DatabaseType::Solr),
Some(SearchEngineQueryRisk::Dangerous)
);
assert_eq!(
classify_search_engine_query_risk("GET /admin/cores?action=STATUS", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
assert_eq!(
classify_search_engine_query_risk("GET /admin/info/system", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
// Replication commands mutate over GET too; only the read whitelist stays safe.
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=details", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=restorestatus", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=filelist", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication", DatabaseType::Solr),
Some(SearchEngineQueryRisk::ReadOnly)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=disablereplication", DatabaseType::Solr),
Some(SearchEngineQueryRisk::Dangerous)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=enablereplication", DatabaseType::Solr),
Some(SearchEngineQueryRisk::Dangerous)
);
assert_eq!(
classify_search_engine_query_risk("GET /mycore/replication?command=fetchindex", DatabaseType::Solr),
Some(SearchEngineQueryRisk::Dangerous)
);
// Core/schema/admin operations are dangerous.
assert_eq!(
classify_search_engine_query_risk("POST /admin/cores?action=CREATE&name=x", DatabaseType::Solr),
Some(SearchEngineQueryRisk::Dangerous)
);
assert_eq!(
classify_search_engine_query_risk(
"POST /mycore/schema\n{\"add-field\":{\"name\":\"x\",\"type\":\"string\"}}",
DatabaseType::Solr
),
Some(SearchEngineQueryRisk::Dangerous)
);
// Non-Solr types never take this classifier.
assert!(classify_search_engine_query_risk("POST /mycore/update", DatabaseType::Postgres).is_none());
}
#[test]
fn excludes_victoriametrics_from_sql_query_paths() {
assert!(!supports_sql_query(DatabaseType::VictoriaMetrics));
assert!(supports_sql_query(DatabaseType::Postgres));
assert_eq!(
check_read_only(
r#"{__name__="dbx_issue_5352_temperature_celsius"}[5m]"#,
"VictoriaMetrics",
DatabaseType::VictoriaMetrics,
),
Ok(())
);
}
#[test]
fn mysql_explicit_transaction_openers_are_recognized() {
for sql in [
"BEGIN",
"begin",
"BEGIN;",
" BEGIN ",
"BEGIN WORK",
"BEGIN WORK;",
"-- open a transaction\nBEGIN",
"/* keep */ START TRANSACTION",
"START TRANSACTION",
"start transaction;",
"START TRANSACTION READ ONLY",
"START TRANSACTION READ WRITE",
"START TRANSACTION WITH CONSISTENT SNAPSHOT",
] {
assert!(mysql_statement_opens_explicit_transaction(sql), "expected opener: {sql}");
}
}
#[test]
fn mysql_non_openers_are_not_treated_as_explicit_transactions() {
for sql in [
"",
" ",
";",
"SELECT 1",
"COMMIT",
"ROLLBACK",
"BEGIN;\nUPDATE t SET a = 1;",
"BEGIN\nDECLARE x INT;\nEND",
"BEGIN END",
"START REPLICA",
"SET autocommit = 0",
"SELECT 'BEGIN' FROM t",
] {
assert!(!mysql_statement_opens_explicit_transaction(sql), "expected non-opener: {sql}");
}
}
}