Require two blank lines between SQL statements

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)

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