Support DDL SQL syntax validation

Author: wayyoungboy

src/common/diag_cmd.py

@@ -1369,7 +1369,7 @@ class ObdiagToolSqlSyntaxCommand(ObdiagOriginCommand):
     def __init__(self):
         super(ObdiagToolSqlSyntaxCommand, self).__init__(
             'sql_syntax',
-            'obdiag tool sql_syntax. Validate SQL against a live OceanBase instance using EXPLAIN (no execution of the original statement)',
+            'obdiag tool sql_syntax. Validate SQL against a live OceanBase instance using EXPLAIN/PREPARE (no execution of the original statement)',
         )
         self.parser.add_option('--sql', type='string', help='SQL statement to validate (single statement only)')
         self.parser.add_option('--env', action='append', type='string', help='Connection override: --env key=value (host, port, user, password/pwd, database/db)')

src/handler/agent/toolsets/obdiag.py

@@ -100,7 +100,7 @@ def truncate_for_agent(
     "rca_run": "执行根因分析（RCA）",
     "rca_list": "列出 RCA 场景",
     "tool_io_performance": "检查节点磁盘 IO",
-    "tool_sql_syntax": "用 EXPLAIN 验证 SQL 语法/语义（不执行）",
+    "tool_sql_syntax": "用 EXPLAIN/PREPARE 验证 SQL 语法/语义（不执行）",
     "list_obdiag_clusters": "列出 obdiag 集群配置",
     "show_current_cluster": "显示当前会话集群与配置路径",
     "db_query": "对集群执行只读 SQL",
@@ -771,7 +771,7 @@ def tool_sql_syntax(
         env: Optional[List[str]] = None,
         cluster_config_path: Optional[str] = None,
     ) -> str:
-        """Validate SQL syntax/semantics using EXPLAIN — does not execute the statement (obdiag tool sql_syntax).
+        """Validate SQL syntax/semantics using EXPLAIN/PREPARE — does not execute the statement (obdiag tool sql_syntax).
 
         Args:
             sql: Single SQL statement to check

src/handler/tools/sql_syntax_handler.py

@@ -14,7 +14,7 @@
 @time: 2026/03/27
 @file: sql_syntax_handler.py
 @desc: Validate SQL syntax/semantics against a live OceanBase instance
-       using EXPLAIN — without executing the SQL.
+       using EXPLAIN or PREPARE — without executing the SQL.
        See https://github.com/oceanbase/obdiag/issues/1181
 """
 
@@ -45,6 +45,32 @@ def normalize_sql_for_syntax_check(sql):
     return s, None
 
 
+def strip_leading_block_comments(sql):
+    pos = 0
+    length = len(sql)
+    while pos < length:
+        while pos < length and sql[pos].isspace():
+            pos += 1
+        if not sql.startswith("/*", pos):
+            break
+        end = sql.find("*/", pos + 2)
+        if end == -1:
+            return ""
+        pos = end + 2
+    return sql[pos:]
+
+
+def is_ddl_statement(sql):
+    matched = re.match(r'(\w+)', strip_leading_block_comments(sql))
+    if not matched:
+        return False
+    return matched.group(1).upper() in ("ALTER", "CREATE", "DROP", "RENAME", "TRUNCATE")
+
+
+def quote_sql_literal(value):
+    return str(value).replace("\\", "\\\\").replace("'", "''")
+
+
 class SqlSyntaxHandler:
     def __init__(self, context):
         self.context = context
@@ -131,33 +157,62 @@ def _resolve_connection(self):
         return host, int(port), user, password, database
 
     def _check_syntax(self, connector, sql):
-        """Run EXPLAIN against the SQL and interpret the result."""
+        if is_ddl_statement(sql):
+            return self._check_ddl_syntax(connector, sql)
+        return self._check_explain_syntax(connector, sql)
+
+    def _check_explain_syntax(self, connector, sql):
+        """Run EXPLAIN against DML SQL and interpret the result."""
         explain_sql = "EXPLAIN {0}".format(sql)
         self.stdio.verbose("[sql-syntax] exec: {0}".format(explain_sql))
 
         try:
             connector.execute_sql(explain_sql)
             self.stdio.print("Result: VALID")
             return ObdiagResult(ObdiagResult.SUCCESS_CODE, data={"result": "VALID", "sql": sql})
-
         except mysql.Error as e:
-            error_code = e.args[0] if e.args else None
-            error_msg = e.args[1] if len(e.args) > 1 else str(e)
-
-            if error_code == 1064:
-                self.stdio.print("Result: SYNTAX ERROR")
-                self.stdio.print("Detail: {0}".format(error_msg))
-                return ObdiagResult(
-                    ObdiagResult.SUCCESS_CODE,
-                    data={"result": "SYNTAX_ERROR", "error_code": error_code, "detail": error_msg},
-                )
-            else:
-                self.stdio.print("Result: VALID (syntax OK, but semantic error [{0}]: {1})".format(error_code, error_msg))
-                return ObdiagResult(
-                    ObdiagResult.SUCCESS_CODE,
-                    data={"result": "SEMANTIC_ERROR", "error_code": error_code, "detail": error_msg},
-                )
+            return self._handle_mysql_syntax_error(e)
+        except Exception as e:
+            self.stdio.error("Unexpected error during SQL syntax check: {0}".format(e))
+            return ObdiagResult(ObdiagResult.SERVER_ERROR_CODE, error_data=str(e))
 
+    def _check_ddl_syntax(self, connector, sql):
+        """Use PREPARE to validate DDL syntax without executing the DDL statement."""
+        stmt_name = "obdiag_sql_syntax_stmt"
+        prepare_sql = "PREPARE {0} FROM '{1}'".format(stmt_name, quote_sql_literal(sql))
+        prepared = False
+        self.stdio.verbose("[sql-syntax] exec: {0}".format(prepare_sql))
+        try:
+            connector.execute_sql(prepare_sql)
+            prepared = True
+            self.stdio.print("Result: VALID")
+            return ObdiagResult(ObdiagResult.SUCCESS_CODE, data={"result": "VALID", "sql": sql, "method": "PREPARE"})
+        except mysql.Error as e:
+            return self._handle_mysql_syntax_error(e)
         except Exception as e:
             self.stdio.error("Unexpected error during SQL syntax check: {0}".format(e))
             return ObdiagResult(ObdiagResult.SERVER_ERROR_CODE, error_data=str(e))
+        finally:
+            if prepared:
+                try:
+                    connector.execute_sql("DEALLOCATE PREPARE {0}".format(stmt_name))
+                except Exception as e:
+                    self.stdio.warn("Failed to deallocate prepared statement {0}: {1}".format(stmt_name, e))
+
+    def _handle_mysql_syntax_error(self, error):
+        error_code = error.args[0] if error.args else None
+        error_msg = error.args[1] if len(error.args) > 1 else str(error)
+
+        if error_code == 1064:
+            self.stdio.print("Result: SYNTAX ERROR")
+            self.stdio.print("Detail: {0}".format(error_msg))
+            return ObdiagResult(
+                ObdiagResult.SUCCESS_CODE,
+                data={"result": "SYNTAX_ERROR", "error_code": error_code, "detail": error_msg},
+            )
+
+        self.stdio.print("Result: VALID (syntax OK, but semantic error [{0}]: {1})".format(error_code, error_msg))
+        return ObdiagResult(
+            ObdiagResult.SUCCESS_CODE,
+            data={"result": "SEMANTIC_ERROR", "error_code": error_code, "detail": error_msg},
+        )

test/handler/tools/test_sql_syntax_handler.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python
+# -*- coding: UTF-8 -*-
+# Copyright (c) 2022 OceanBase
+# OceanBase Diagnostic Tool is licensed under Mulan PSL v2.
+# You can use this software according to the terms and conditions of the Mulan PSL v2.
+# You may obtain a copy of Mulan PSL v2 at:
+#          http://license.coscl.org.cn/MulanPSL2
+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+# See the Mulan PSL v2 for more details.
+
+"""
+@time: 2026/6/8
+@file: test_sql_syntax_handler.py
+@desc:
+"""
+
+import unittest
+from unittest.mock import MagicMock
+
+import pymysql as mysql
+
+from src.common.result_type import ObdiagResult
+from src.handler.tools.sql_syntax_handler import SqlSyntaxHandler, is_ddl_statement, normalize_sql_for_syntax_check, quote_sql_literal
+
+
+class _Stdio:
+    def print(self, *args, **kwargs):
+        pass
+
+    def verbose(self, *args, **kwargs):
+        pass
+
+    def warn(self, *args, **kwargs):
+        pass
+
+    def error(self, *args, **kwargs):
+        pass
+
+
+class TestSqlSyntaxHandler(unittest.TestCase):
+    def setUp(self):
+        context = MagicMock()
+        context.stdio = _Stdio()
+        context.options = {}
+        self.handler = SqlSyntaxHandler(context)
+
+    def test_normalize_accepts_single_statement(self):
+        sql, err = normalize_sql_for_syntax_check("  SELECT 1;; ")
+        self.assertIsNone(err)
+        self.assertEqual(sql, "SELECT 1")
+
+    def test_normalize_rejects_multiple_statements(self):
+        sql, err = normalize_sql_for_syntax_check("SELECT 1; SELECT 2")
+        self.assertIsNotNone(err)
+        self.assertIsNone(sql)
+
+    def test_detects_ddl_statement(self):
+        self.assertTrue(is_ddl_statement("CREATE TABLE t1(id int)"))
+        self.assertTrue(is_ddl_statement("/* comment */ ALTER TABLE t1 ADD c1 int"))
+        self.assertTrue(is_ddl_statement("  /* a */ /* b */ DROP TABLE t1"))
+        self.assertFalse(is_ddl_statement("SELECT * FROM t1"))
+        self.assertFalse(is_ddl_statement("/* unfinished comment CREATE TABLE t1(id int)"))
+
+    def test_quote_sql_literal_escapes_backslash_and_quote(self):
+        self.assertEqual(quote_sql_literal("CREATE TABLE `a\\b` (c varchar(10) default 'x')"), "CREATE TABLE `a\\\\b` (c varchar(10) default ''x'')")
+
+    def test_dml_uses_explain(self):
+        connector = MagicMock()
+        result = self.handler._check_syntax(connector, "SELECT * FROM t1")
+        self.assertEqual(result.code, ObdiagResult.SUCCESS_CODE)
+        self.assertEqual(result.data["result"], "VALID")
+        connector.execute_sql.assert_called_once_with("EXPLAIN SELECT * FROM t1")
+
+    def test_ddl_uses_prepare_without_execute(self):
+        connector = MagicMock()
+        result = self.handler._check_syntax(connector, "CREATE TABLE t1(id int)")
+        self.assertEqual(result.code, ObdiagResult.SUCCESS_CODE)
+        self.assertEqual(result.data["result"], "VALID")
+        self.assertEqual(result.data["method"], "PREPARE")
+        calls = [call.args[0] for call in connector.execute_sql.call_args_list]
+        self.assertEqual(calls[0], "PREPARE obdiag_sql_syntax_stmt FROM 'CREATE TABLE t1(id int)'")
+        self.assertEqual(calls[1], "DEALLOCATE PREPARE obdiag_sql_syntax_stmt")
+
+    def test_ddl_syntax_error_does_not_deallocate_unprepared_statement(self):
+        connector = MagicMock()
+        connector.execute_sql.side_effect = mysql.Error(1064, "syntax error")
+        result = self.handler._check_syntax(connector, "CREATE TABLE t1(")
+        self.assertEqual(result.data["result"], "SYNTAX_ERROR")
+        connector.execute_sql.assert_called_once()
+
+    def test_non_1064_error_is_reported_as_semantic_error(self):
+        connector = MagicMock()
+        connector.execute_sql.side_effect = mysql.Error(1146, "table does not exist")
+        result = self.handler._check_syntax(connector, "SELECT * FROM missing_table")
+        self.assertEqual(result.data["result"], "SEMANTIC_ERROR")
+        self.assertEqual(result.data["error_code"], 1146)
+
+
+if __name__ == '__main__':
+    unittest.main()