Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 6 additions & 9 deletions src/core/schema_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@

logger = get_logger(__name__)

# A line that starts with one of these is a table constraint, not a column.
CONSTRAINT_LINE_RE = re.compile(r"(PRIMARY\s+KEY|FOREIGN\s+KEY|CONSTRAINT|UNIQUE|CHECK|EXCLUDE)\b")


class SchemaParser:
"""Parse CREATE TABLE and ALTER TABLE statements from SQL text.
Expand Down Expand Up @@ -193,15 +196,9 @@ def _parse_column_line_robust(self, line: str) -> dict | None:

# Skip pure constraint lines
upper = line.upper().strip()
skip_patterns = (
"PRIMARY KEY",
"FOREIGN KEY",
"CONSTRAINT",
"UNIQUE",
"CHECK",
"EXCLUDE",
)
if upper.startswith(skip_patterns) and "(" in upper:
# Anchored on a word boundary so a column named unique_code or
# constraint_name is not mistaken for a constraint declaration.
if CONSTRAINT_LINE_RE.match(upper) and "(" in upper:
return None

# Try to extract column name and type
Expand Down
34 changes: 34 additions & 0 deletions tests/test_schema_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,40 @@ def test_table_level_primary_key(self, parser):
assert len(pks) == 2
assert {c["name"] for c in pks} == {"a", "b"}

def test_column_named_like_constraint_keyword(self, parser):
"""A column whose name starts with a constraint keyword is a column.

unique_code and constraint_name used to be dropped because the
constraint check compared raw prefixes. The designer then showed a
table with fewer columns than the SQL declares.
"""
sql = """
CREATE TABLE t (
id SERIAL PRIMARY KEY,
unique_code VARCHAR(10),
constraint_name VARCHAR(50),
check_flag BOOLEAN,
excluded_at TIMESTAMP
);
"""
tables, _ = parser.parse_sql(sql)
names = [c["name"] for c in tables[0]["columns"]]
assert names == ["id", "unique_code", "constraint_name", "check_flag", "excluded_at"]

def test_table_level_unique_constraint_still_skipped(self, parser):
"""The keyword fix must not start parsing real constraint lines."""
sql = """
CREATE TABLE t (
id SERIAL PRIMARY KEY,
email VARCHAR(200),
UNIQUE (email),
CONSTRAINT chk_len CHECK (length(email) > 3)
);
"""
tables, _ = parser.parse_sql(sql)
names = [c["name"] for c in tables[0]["columns"]]
assert names == ["id", "email"]

def test_not_null(self, parser):
sql = "CREATE TABLE t (name VARCHAR(100) NOT NULL);"
tables, _ = parser.parse_sql(sql)
Expand Down
Loading