diff --git a/sqlit/domains/query/app/multi_statement.py b/sqlit/domains/query/app/multi_statement.py index 381a97f6..26077d32 100644 --- a/sqlit/domains/query/app/multi_statement.py +++ b/sqlit/domains/query/app/multi_statement.py @@ -9,13 +9,17 @@ from __future__ import annotations import re +from collections.abc import Iterator from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Iterator +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .query_service import NonQueryResult, QueryResult +_DOUBLE_BLANK_LINE_RE = re.compile(r"\r?\n[ \t]*\r?\n[ \t]*\r?\n") + + def _iter_sql_chars(sql: str) -> Iterator[tuple[int, str, bool]]: """Iterate through SQL characters, tracking string literal context. @@ -94,10 +98,7 @@ def _iter_sql_chars(sql: str) -> Iterator[tuple[int, str, bool]]: def _has_semicolon_outside_strings(sql: str) -> bool: """Check if SQL has semicolons outside of string literals.""" - for _, char, outside in _iter_sql_chars(sql): - if char == ";" and outside: - return True - return False + return any(char == ";" and outside for _, char, outside in _iter_sql_chars(sql)) def _split_by_semicolons(sql: str) -> list[str]: @@ -123,10 +124,11 @@ def _split_by_semicolons(sql: str) -> list[str]: def _split_by_blank_lines(sql: str) -> list[str]: - """Split SQL by blank lines, respecting string literals. + """Split SQL by two consecutive blank lines, respecting string literals. A blank line is defined as a line containing only whitespace. - This is triggered when there are no semicolons in the query. + A single blank line remains ordinary SQL formatting; the split occurs only + after a second consecutive blank line when there are no semicolons. """ statements = [] current: list[str] = [] @@ -138,24 +140,21 @@ def _split_by_blank_lines(sql: str) -> list[str]: line_content = sql[line_start:idx] current_line_empty = not line_content.strip() - if current_line_empty and prev_line_empty: - # Consecutive blank lines, skip - pass - elif current_line_empty and current: - # Blank line after content - split here + if current_line_empty and prev_line_empty and current: + # The second consecutive blank line is the boundary. stmt = "".join(current).strip() if stmt: statements.append(stmt) current = [] else: - # Regular newline, keep it + # Keep regular newlines and the first blank line as formatting. current.append(char) prev_line_empty = current_line_empty line_start = idx + 1 else: current.append(char) - if char not in " \t\n": + if char not in " \t\r\n": prev_line_empty = False # Don't forget the last statement @@ -182,7 +181,7 @@ def _get_statement_ranges(sql: str) -> list[tuple[str, int, int]]: Splitting strategy (matches split_statements): 1. If query contains semicolons (outside strings) → split by semicolons - 2. If no semicolons but has blank lines → split by blank lines + 2. If no semicolons but has two consecutive blank lines → split there 3. Otherwise → return as single statement Returns: @@ -206,8 +205,8 @@ def _get_statement_ranges(sql: str) -> list[tuple[str, int, int]]: _append_statement_range(ranges, sql, stmt_start, len(sql)) return ranges - # Strategy 2: If blank lines exist, use blank line splitting with tracking - if re.search(r"\n\s*\n", sql): + # Strategy 2: Two consecutive blank lines delimit statements. + if _DOUBLE_BLANK_LINE_RE.search(sql): stmt_start = 0 line_start = 0 prev_line_empty = False @@ -218,16 +217,13 @@ def _get_statement_ranges(sql: str) -> list[tuple[str, int, int]]: current_line_empty = not line_content.strip() if current_line_empty and prev_line_empty: - # Consecutive blank lines, skip - pass - elif current_line_empty: - # Blank line after content - this is a statement boundary + # The second consecutive blank line is the boundary. _append_statement_range(ranges, sql, stmt_start, idx) stmt_start = idx + 1 prev_line_empty = current_line_empty line_start = idx + 1 - elif char not in " \t\n": + elif char not in " \t\r\n": prev_line_empty = False _append_statement_range(ranges, sql, stmt_start, len(sql)) @@ -312,12 +308,12 @@ def split_statements(sql: str) -> list[str]: Splitting strategy: 1. If query contains semicolons (outside strings) → split by semicolons - 2. If no semicolons but has blank lines → split by blank lines + 2. If no semicolons but has two consecutive blank lines → split there 3. Otherwise → return as single statement Handles: - Multiple statements separated by semicolons - - Multiple statements separated by blank lines (when no semicolons) + - Multiple statements separated by two blank lines (when no semicolons) - Semicolons/blank lines inside string literals (preserved) - Empty statements (filtered out) - Trailing semicolons @@ -335,9 +331,9 @@ def split_statements(sql: str) -> list[str]: if _has_semicolon_outside_strings(sql): return _split_by_semicolons(sql) - # Strategy 2: If blank lines exist, use blank line splitting - # A blank line is two consecutive newlines (possibly with whitespace between) - if re.search(r"\n\s*\n", sql): + # Strategy 2: Two blank lines are three newlines, allowing one blank line + # to remain ordinary formatting inside a statement. + if _DOUBLE_BLANK_LINE_RE.search(sql): return _split_by_blank_lines(sql) # Strategy 3: Single statement @@ -347,7 +343,7 @@ def split_statements(sql: str) -> list[str]: def normalize_for_execution(sql: str) -> str: """Normalize SQL for database execution. - Converts blank-line-separated statements to semicolon-separated, + Converts double-blank-line-separated statements to semicolon-separated, since databases expect semicolons between statements. Args: @@ -363,8 +359,8 @@ def normalize_for_execution(sql: str) -> str: if _has_semicolon_outside_strings(sql): return sql - # If has blank lines, split and rejoin with semicolons - if re.search(r"\n\s*\n", sql): + # If it has two consecutive blank lines, split and rejoin with semicolons. + if _DOUBLE_BLANK_LINE_RE.search(sql): statements = _split_by_blank_lines(sql) if len(statements) > 1: return "; ".join(statements) diff --git a/tests/unit/test_multi_statement.py b/tests/unit/test_multi_statement.py index 617abb2c..7f0e904b 100644 --- a/tests/unit/test_multi_statement.py +++ b/tests/unit/test_multi_statement.py @@ -6,8 +6,6 @@ from __future__ import annotations -import pytest - class TestStatementSplitting: """Tests for splitting multi-statement queries.""" @@ -306,16 +304,18 @@ def test_preserves_statement_text_in_results(self): class TestBlankLineSplitting: - """Tests for splitting statements by blank lines when no semicolons.""" + """Tests for splitting statements by two blank lines when no semicolons.""" - def test_splits_by_blank_line_when_no_semicolons(self): - """Should split by blank lines when query has no semicolons.""" + def test_splits_by_two_blank_lines_when_no_semicolons(self): + """Two consecutive blank lines split when the query has no semicolons.""" from sqlit.domains.query.app.multi_statement import split_statements query = """SELECT * FROM vikings + SELECT * FROM ships + SELECT * FROM weapons""" statements = split_statements(query) @@ -351,6 +351,18 @@ def test_multiline_statement_stays_together(self): assert "JOIN" in statements[0] assert "WHERE" in statements[0] + def test_single_blank_line_inside_statement_stays_together(self): + """Issue #281: one blank line is ordinary formatting, not a boundary.""" + from sqlit.domains.query.app.multi_statement import split_statements + + query = """SELECT * +FROM schema_name.table_name +WHERE id < 1000 + +LIMIT 10""" + + assert split_statements(query) == [query] + def test_blank_line_with_multiline_statements(self): """Should split by blank lines even with multi-line statements.""" from sqlit.domains.query.app.multi_statement import split_statements @@ -359,6 +371,7 @@ def test_blank_line_with_multiline_statements(self): FROM vikings v WHERE v.id = 1 + SELECT s.name FROM ships s WHERE s.id = 2""" @@ -394,23 +407,37 @@ def test_single_newline_does_not_split(self): assert len(statements) == 1 - def test_blank_line_with_whitespace_only(self): - """Line with only whitespace should count as blank line.""" + def test_two_blank_lines_with_whitespace_only(self): + """Two whitespace-only blank lines should count as a boundary.""" from sqlit.domains.query.app.multi_statement import split_statements - query = "SELECT 1\n \nSELECT 2" + query = "SELECT 1\n \n\t\nSELECT 2" statements = split_statements(query) assert len(statements) == 2 + def test_two_crlf_blank_lines_are_a_boundary(self): + """Windows line endings preserve the double-blank-line separator.""" + from sqlit.domains.query.app.multi_statement import ( + normalize_for_execution, + split_statements, + ) + + query = "SELECT 1\r\n\r\n\r\nSELECT 2" + + assert split_statements(query) == ["SELECT 1", "SELECT 2"] + assert normalize_for_execution(query) == "SELECT 1; SELECT 2" + def test_preserves_strings_with_newlines_in_blank_line_mode(self): """Should not split on blank lines inside string literals.""" from sqlit.domains.query.app.multi_statement import split_statements query = """SELECT 'line1 + line2' AS text + SELECT 'other'""" statements = split_statements(query) @@ -422,14 +449,16 @@ def test_preserves_strings_with_newlines_in_blank_line_mode(self): class TestNormalizeSqlForExecution: """Tests for normalizing SQL before execution.""" - def test_adds_semicolons_between_blank_line_statements(self): - """Blank-line-separated statements should be joined with semicolons for execution.""" + def test_adds_semicolons_between_double_blank_line_statements(self): + """Double-blank-line statements are joined with semicolons for execution.""" from sqlit.domains.query.app.multi_statement import normalize_for_execution query = """SELECT * FROM vikings + SELECT * FROM ships + SELECT * FROM weapons""" normalized = normalize_for_execution(query) @@ -440,6 +469,14 @@ def test_adds_semicolons_between_blank_line_statements(self): assert "ships" in normalized assert "weapons" in normalized + def test_preserves_single_blank_line(self): + """A single blank line remains part of one statement.""" + from sqlit.domains.query.app.multi_statement import normalize_for_execution + + query = "SELECT * FROM vikings\n\nLIMIT 10" + + assert normalize_for_execution(query) == query + def test_preserves_semicolon_separated_statements(self): """Already semicolon-separated statements should stay as-is.""" from sqlit.domains.query.app.multi_statement import normalize_for_execution