From a236127903af168d7da4ad712caa0f85633d4f43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Furga=C5=82a?= <83299832+00200200@users.noreply.github.com> Date: Sat, 24 Oct 2026 16:49:35 +0200 Subject: [PATCH] Recognize TOP as a keyword while preserving qualified names --- CHANGELOG | 3 +- sqlparse/keywords.py | 2 ++ sqlparse/lexer.py | 18 ++++++++--- tests/test_parse.py | 71 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 5 deletions(-) diff --git a/CHANGELOG b/CHANGELOG index 44d5938e..f22092a8 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,7 +1,8 @@ Development Version ------------------- -Nothing yet. +* Recognize TOP as a keyword, including parenthesized row limits, so it is + parsed and formatted consistently in SQL Server queries (issue703). Release 0.6.0 (Aug 13, 2026) diff --git a/sqlparse/keywords.py b/sqlparse/keywords.py index dd6e5d15..fcb2f07d 100644 --- a/sqlparse/keywords.py +++ b/sqlparse/keywords.py @@ -157,6 +157,7 @@ def find_delimited_spans(text): # FIXME(atronah): never match, # because `re.match` doesn't work with look-behind regexp feature (r'(?<=\.)[A-ZÀ-Ü]\w*', tokens.Name), # .'Name' + (r'TOP\b(?=\s*\()', tokens.Keyword), # Parenthesized row limit, not a function. (r'[A-ZÀ-Ü]\w*(?=\()', tokens.Name), # side effect: change kw to func (r'-?0x[\dA-F]+', tokens.Number.Hexadecimal), (r'-?\d+(\.\d+)?E-?\d+', tokens.Number.Float), @@ -667,6 +668,7 @@ def find_delimited_spans(text): 'TIMEZONE_MINUTE': tokens.Keyword, 'TO': tokens.Keyword, 'TOAST': tokens.Keyword, + 'TOP': tokens.Keyword, 'TRAILING': tokens.Keyword, 'TRANSATION': tokens.Keyword, 'TRANSACTIONS_COMMITTED': tokens.Keyword, diff --git a/sqlparse/lexer.py b/sqlparse/lexer.py index 5d4a5f97..b9d4a57d 100644 --- a/sqlparse/lexer.py +++ b/sqlparse/lexer.py @@ -136,6 +136,7 @@ def get_tokens(self, text, encoding=None): span_openers = delimited_spans.openers iterable = enumerate(text) + after_dot = False for pos, char in iterable: # Only positions the lexer actually reaches may open a # dollar-quoted literal or a multiline comment; a delimiter @@ -145,6 +146,8 @@ def get_tokens(self, text, encoding=None): resolved = delimited_spans.resolve(pos) if resolved is not None: end, ttype = resolved + if ttype not in tokens.Comment: + after_dot = False yield ttype, text[pos:end] consume(iterable, end - pos - 1) continue @@ -154,14 +157,21 @@ def get_tokens(self, text, encoding=None): if not m: continue - elif isinstance(action, tokens._TokenType): - yield action, m.group() - elif action is keywords.PROCESS_AS_KEYWORD: - yield self.is_keyword(m.group()) + value = m.group() + if action is keywords.PROCESS_AS_KEYWORD: + action, value = self.is_keyword(value) + if isinstance(action, tokens._TokenType): + # Qualified TOP stays a name across whitespace/comments. + if after_dot and action is tokens.Keyword and value.upper() == 'TOP': + action = tokens.Name + if action not in tokens.Whitespace and action not in tokens.Comment: + after_dot = action is tokens.Punctuation and value == '.' + yield action, value consume(iterable, m.end() - pos - 1) break else: + after_dot = False yield tokens.Error, char diff --git a/tests/test_parse.py b/tests/test_parse.py index 67168410..d4672b92 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -114,6 +114,77 @@ def test_parse_keyword_like_identifier(): assert isinstance(t[0], sql.Identifier) +@pytest.mark.parametrize('top', ['TOP', 'top', 'ToP']) +@pytest.mark.parametrize('limit', [' 10', ' (10)', '(10)', '\n(10)']) +def test_parse_top_keyword(top, limit): + # see issue703 + statement = f'select {top}{limit} "name", "id" from some_table' + parsed = sqlparse.parse(statement)[0] + assert parsed.tokens[2].match(T.Keyword, 'TOP') + assert str(parsed) == statement + assert sqlparse.format(statement, keyword_case='upper') == ( + f'SELECT TOP{limit} "name", "id" FROM some_table') + + +@pytest.mark.parametrize('name', [ + '[top]', '"top"', '`top`', 'top_count', 'top.value', +]) +def test_parse_top_identifier(name): + statement = f'select {name} from some_table' + parsed = sqlparse.parse(statement)[0] + assert isinstance(parsed.tokens[2], sql.Identifier) + assert str(parsed.tokens[2]) == name + assert sqlparse.format(statement, keyword_case='upper') == ( + f'SELECT {name} FROM some_table') + + +@pytest.mark.parametrize('qualifier', ['schema', '"schema"']) +@pytest.mark.parametrize('separator', ['', ' ', '\n\t']) +@pytest.mark.parametrize('suffix', ['', '(1)']) +def test_parse_qualified_top(qualifier, separator, suffix): + name = f'{qualifier}.{separator}top{suffix}' + statement = f'select {name} from some_table' + parsed = sqlparse.parse(statement)[0] + identifier = parsed.tokens[2] + assert isinstance(identifier, sql.Identifier) + assert str(identifier) == name + assert identifier.get_real_name() == 'top' + assert identifier.get_parent_name() == 'schema' + functions = [token for token in identifier.tokens + if isinstance(token, sql.Function)] + if suffix: + assert len(functions) == 1 + assert functions[0].get_name() == 'top' + assert [str(token) for token in functions[0].get_parameters()] == ['1'] + else: + assert not functions + assert str(parsed) == statement + assert sqlparse.format(statement, keyword_case='upper') == ( + f'SELECT {name} FROM some_table') + + +@pytest.mark.parametrize('qualifier', ['schema', '"schema"']) +@pytest.mark.parametrize('comment', ['/* comment */', '-- comment\n']) +@pytest.mark.parametrize('suffix', ['', '(1)']) +def test_parse_qualified_top_with_comment(qualifier, comment, suffix): + name = f'{qualifier}. {comment} top{suffix}' + statement = f'select {name} from some_table' + parsed = sqlparse.parse(statement)[0] + identifiers = [token for token in parsed.tokens + if isinstance(token, (sql.Identifier, sql.Function))] + # Comments already separate the qualifier and name into distinct groups. + assert identifiers[0].get_parent_name() == 'schema' + assert identifiers[0].get_real_name() is None + assert identifiers[1].get_real_name() == 'top' + assert identifiers[1].get_parent_name() is None + assert isinstance(identifiers[1], sql.Function if suffix else sql.Identifier) + if suffix: + assert [str(token) for token in identifiers[1].get_parameters()] == ['1'] + assert str(parsed) == statement + assert sqlparse.format(statement, keyword_case='upper') == ( + f'SELECT {name} FROM some_table') + + def test_parse_function_parameter(): # see issue94 t = sqlparse.parse('abs(some_col)')[0].tokens[0].get_parameters()