# Copyright (c) 2024, 2025, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, as # published by the Free Software Foundation. # # This program is designed to work with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, # as designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an # additional permission to link the program and your derivative works # with the separately licensed software that they have either included with # the program or referenced in the documentation. # # Without limiting anything contained in the foregoing, this file, # which is part of MySQL Connector/Python, is also subject to the # Universal FOSS Exception, version 1.0, a copy of which can be found at # http://oss.oracle.com/licenses/universal-foss-exception. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. # See the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA """Classes and methods utilized to work with MySQL Scripts.""" import re import unicodedata from collections import deque from typing import Deque, Generator, Optional from .errors import InterfaceError from .types import MySQLScriptPartition DEFAULT_DELIMITER = b";" """The default delimiter of MySQL Client and the only one recognized by the MySQL server protocol.""" DELIMITER_RESERVED_SYMBOLS = { "$": rb"\$", "^": rb"\^", "?": rb"\?", "(": rb"\(", ")": rb"\)", "[": rb"\[", "]": rb"\]", "{": rb"\{", "}": rb"\}", ".": rb"\.", "|": rb"\|", "+": rb"\+", "-": rb"\-", "*": rb"\*", } """Symbols with a special meaning in regular expression contexts.""" DELIMITER_PATTERN: re.Pattern = re.compile( rb"""(delimiter\s+)(?=(?:[^"'`]*(?:"[^"]*"|'[^']*'|`[^`]*`))*[^"'`]*$)""", flags=re.IGNORECASE | re.MULTILINE, ) """Regular expression pattern recognizing the delimiter command.""" class MySQLScriptSplitter: """Breaks a MySQL script into single statements. It strips custom delimiters and comments along the way, except for comments representing a MySQL extension or optimizer hint. """ _regex_sql_split_stmts = b"""(?=(?:[^"'`]*(?:"[^"]*"|'[^']*'|`[^`]*`))*[^"'`]*$)""" def __init__(self, sql_script: bytes) -> None: """Constructor.""" self._code = sql_script self._single_stmts: Optional[list[bytes]] = None self._mappable_stmts: Optional[list[bytes]] = None self._re_sql_split_stmts: dict[bytes, re.Pattern] = {} def _split_statement(self, code: bytes, delimiter: bytes) -> list[bytes]: """Split code context by delimiter.""" snippets = [] if delimiter not in self._re_sql_split_stmts: if b"\\" in delimiter: raise InterfaceError( "The backslash (\\) character is not a valid delimiter." ) delimiter_pattern = [ DELIMITER_RESERVED_SYMBOLS.get(char, char.encode()) for char in delimiter.decode() ] self._re_sql_split_stmts[delimiter] = re.compile( b"".join(delimiter_pattern) + self._regex_sql_split_stmts ) for snippet in self._re_sql_split_stmts[delimiter].split(code): snippet_strip = snippet.strip() if snippet_strip: snippets.append(snippet_strip) return snippets @staticmethod def is_white_space_char(char: int) -> bool: """Validates whether `char` is a white-space character or not.""" return unicodedata.category(chr(char))[0] in {"Z"} @staticmethod def is_control_char(char: int) -> bool: """Validates whether `char` is a control character or not.""" return unicodedata.category(chr(char))[0] in {"C"} @staticmethod def split_by_control_char_or_white_space(string: bytes) -> list[bytes]: """Split `string` by any control character or whitespace.""" return re.split(rb"[\s\x00-\x1f\x7f-\x9f]", string) @staticmethod def has_delimiter(code: bytes) -> bool: """Validates whether `code` has the delimiter command pattern or not.""" return re.search(DELIMITER_PATTERN, code) is not None @staticmethod def remove_comments(code: bytes) -> bytes: """Remove MySQL comments which include `--`-style, `#`-style and `C`-style comments. A `--`-style comment spans from `--` to the end of the line. It requires the second dash to be followed by at least one whitespace or control character (such as a space, tab, newline, and so on). A `#`-style comment spans from `#` to the end of the line. A C-style comment spans from a `/*` sequence to the following `*/` sequence, as in the C programming language. This syntax enables a comment to extend over multiple lines because the beginning and closing sequences need not be on the same line. **NOTE: Only C-style comments representing MySQL extensions or optimizer hints are preserved**. E.g., ``` /*! MySQL-specific code */ /*+ MySQL-specific code */ ``` *For Reference Manual- MySQL Comments*, see https://dev.mysql.com/doc/refman/en/comments.html. """ def is_dash_style(b_str: bytes, b_char: int) -> bool: return b_str == b"--" and ( MySQLScriptSplitter.is_control_char(b_char) or MySQLScriptSplitter.is_white_space_char(b_char) ) def is_hash_style(b_str: bytes) -> bool: return b_str == b"#" def is_c_style(b_str: bytes, b_char: int) -> bool: return b_str == b"/*" and b_char not in {ord("!"), ord("+")} buf = bytearray(b"") i, literal_ctx = 0, None line_break, single_quote, double_quote = ord("\n"), ord("'"), ord('"') while i < len(code): if literal_ctx is None: style = None if is_dash_style(buf[-2:], code[i]): style = "--" elif is_hash_style(buf[-1:]): style = "#" elif is_c_style(buf[-2:], code[i]): style = "/*" if style is not None: if style in ("--", "#"): while i < len(code) and code[i] != line_break: i += 1 else: while i + 1 < len(code) and code[i : i + 2] != b"*/": i += 1 i += 2 for _ in range(len(style)): buf.pop() while buf and ( MySQLScriptSplitter.is_control_char(buf[-1]) or MySQLScriptSplitter.is_white_space_char(buf[-1]) ): buf.pop() continue if literal_ctx is None and code[i] in [single_quote, double_quote]: literal_ctx = code[i] elif literal_ctx is not None and code[i] in {literal_ctx, line_break}: literal_ctx = None buf.append(code[i]) i += 1 return bytes(buf) def split_script(self, remove_comments: bool = True) -> list[bytes]: """Splits the given script text into a sequence of individual statements. The word DELIMITER and any of its lower and upper case combinations such as delimiter, DeLiMiter, etc., are considered reserved words by the connector. Users must quote these when included in multi statements for other purposes different from declaring an actual statement delimiter; e.g., as names for tables, columns, variables, in comments, etc. ``` CREATE TABLE `delimiter` (begin INT, end INT); -- I am a `DELimiTer` comment ``` If they are not quoted, the statement-mapping will not produce the expected experience. See https://dev.mysql.com/doc/refman/8.0/en/keywords.html to know more about quoting a reserved word. *Note that comments are always ignored as they are not considered to be part of statements, with one exeception; **C-style comments representing MySQL extensions or optimizer hints are preserved***. """ # If it was already computed, then skip computation and use the cache if self._single_stmts is not None: return self._single_stmts # initialize variables self._single_stmts = [] delimiter = DEFAULT_DELIMITER buf: list[bytes] = [] prev = b"" # remove comments if remove_comments: code = MySQLScriptSplitter.remove_comments(code=self._code) else: code = self._code # let's split the script by `delimiter pattern` - the pattern is also # included in the returned list. for curr in re.split(pattern=DELIMITER_PATTERN, string=code): # Checking if the previous substring is a "switch of context # (delimiter)" point. if re.search(DELIMITER_PATTERN, prev): # The next delimiter must be the sequence of chars until # reaching a control char or whitespace next_delimiter = self.split_by_control_char_or_white_space(curr)[0] # We shall remove the delimiter command from the code buf.pop() # At this point buf includes all the code where `delimiter` applies. self._single_stmts.extend( self._split_statement(code=b" ".join(buf), delimiter=delimiter) ) # From the current substring, let's take everything but the # "next delimiter" portion. Also, let's update the delimiter delimiter, buf = next_delimiter, [curr[len(next_delimiter) :]] else: # Let's accumulate buf.append(curr) # track the previous substring prev = curr # Ensure there are no loose ends if buf: self._single_stmts.extend( self._split_statement(code=b" ".join(buf), delimiter=delimiter) ) return self._single_stmts def __repr__(self) -> str: return self._code.decode("utf-8") def split_multi_statement( sql_code: bytes, map_results: bool = False, ) -> Generator[MySQLScriptPartition, None, None]: """Breaks a MySQL script into sub-scripts. If the given script uses `DELIMITER` statements (which are not recognized by MySQL Server), the connector will parse such statements to remove them from the script and substitute delimiters as needed. This pre-processing may cause a performance hit when using long scripts. Note that when enabling `map_results`, the script is expected to use `DELIMITER` statements in order to split the script into multiple query strings. Args: sql_code: MySQL script. map_results: If True, each sub-script is `statement-result` mappable. Returns: A generator of typed dictionaries with keys `single_stmts` and `mappable_stmts`. If mapping disabled and no delimiters detected, it returns a 1-item generator, the field `single_stmts` is an empty list and the `mappable_stmt` field corresponds to the unmodified script, that may be mappable. If mapping disabled and delimiters detected, it returns a 1-item generator, the field `single_stmts` is a list including all the single statements found in the script and the `mappable_stmt` field corresponds to the processed script (delimiters are stripped) that may be mappable. If maping enabled, the script is broken into mappable partitions. It returns an N-item generator (as many items as computed partitions), the field `single_stmts` is a list including all the single statements of the partition and the `mappable_stmt` field corresponds to the sub-script (partition) that is guaranteed to be mappable. Raises: `InterfaceError` if an invalid delimiter string is found. """ if not MySQLScriptSplitter.has_delimiter(sql_code) and not map_results: # For those users executing single statements or scripts with no delimiters, # they can get a performance boost by bypassing the multi statement splitter. # Simply wrap the multi statement up (so it can be processed correctly # downstream) and return it as it is. yield MySQLScriptPartition(single_stmts=deque([]), mappable_stmt=sql_code) return tok = MySQLScriptSplitter(sql_script=sql_code) # The splitter splits the sql code into many single statements # while also getting rid of the delimiters (if any). stmts = tok.split_script() # if there are not statements to execute if not stmts: # Simply wrap the multi statement up (so it can be processed correctly # downstream). yield MySQLScriptPartition(single_stmts=deque([b""]), mappable_stmt=b"") return if not map_results: # group single statements into a unique and possibly no mappable # multi statement. yield MySQLScriptPartition( single_stmts=deque(stmts), mappable_stmt=b";\n".join(stmts) ) return # group single statements into one or more mappable multi statements. i = 0 partition_ids = (j for j, stmt in enumerate(stmts) if stmt[:5].upper() == b"CALL ") for j in partition_ids: if j > i: yield ( MySQLScriptPartition( mappable_stmt=b";\n".join(stmts[i:j]), single_stmts=deque(stmts[i:j]), ) ) yield MySQLScriptPartition( mappable_stmt=stmts[j], single_stmts=deque([stmts[j]]) ) i = j + 1 if i < len(stmts): yield ( MySQLScriptPartition( mappable_stmt=b";\n".join(stmts[i : len(stmts)]), single_stmts=deque(stmts[i : len(stmts)]), ) ) def get_local_infile_filenames(script: bytes) -> Deque[str]: """Scans the MySQL script looking for `filenames` (one for each `LOCAL INFILE` statement found). Arguments: script: a MySQL script that may include one or more `LOCAL INFILE` statements. Returns: filenames: a list of filenames (one for each `LOCAL INFILE` statement found). An empty list is returned if no matches are found. """ matches = re.findall( pattern=rb"""LOCAL\s+INFILE\s+(["'])((?:\\\1|(?:(?!\1)).)*)(\1)""", string=MySQLScriptSplitter.remove_comments(script), flags=re.IGNORECASE, ) if not matches or len(matches[0]) != 3: return deque([]) # If there is a match, we get ("'", "filename", "'") , that's to say, # the 1st and 3rd entries are the quote symbols, and the 2nd the actual filename. return deque([match[1].decode("utf-8") for match in matches])